本文介绍如何通过事件总线EventBridge消息队列RabbitMQ版的数据推送到函数计算。

注意事项

连接器支持的地域以消息队列RabbitMQ版开服的地域为准。

步骤一:创建自定义事件源

  1. 登录事件总线EventBridge控制台
  2. 在左侧导航栏,选择事件驱动 > 事件源
  3. 在顶部菜单栏,选择地域。
  4. 快速添加自定义事件源区域,单击消息队列RabbitMQ版
  5. 添加自定义事件源面板,输入名称,输入描述,选择RabbitMQ实例,选择Vhost,选择Queue,选择自定义总线,然后单击确定

步骤二:创建事件规则

注意 目标服务和事件规则必须处于同一地域。
  1. 登录事件总线EventBridge控制台
  2. 在左侧导航栏,选择事件驱动 > 事件规则
  3. 在顶部菜单栏,选择地域。
  4. 事件规则页面,选择自定义总线,然后单击创建规则
  5. 在配置向导页面,完成以下操作。
    1. 配置基本信息页面,在名称文本框输入规则名称,在描述文本框输入规则的描述,然后单击下一步
    2. 配置事件模式页面,事件源类型选择自定义事件源事件源选择步骤一创建的自定义事件源,在事件模式内容代码框输入事件模式,然后单击下一步

      如需了解更多信息,请参见事件模式

    3. 配置事件目标页面,配置事件目标,然后单击创建
      说明 1个事件规则最多可以添加5个目标。
      • 服务类型:单击函数计算
      • 服务:选择已创建的服务。
      • 函数:选择已创建的函数。
      • 事件:单击模板

        以下提供变量和模板的示例。

        变量示例:

        {
          "source":"$.source",
          "type":"$.type"
        }

        模板示例:

        The event comes from ${source},event type is ${type}.

        如需了解更多信息,请参见事件内容转换

      • 服务版本和别名:选择服务版本或服务别名。
        • 默认版本:LATEST。
        • 指定版本:选择服务版本。更多信息,请参见版本简介
        • 指定别名:选择服务别名。更多信息,请参见别名简介

步骤三:发布事件

import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
import java.util.HashMap;
import java.util.UUID;

public class ProducerTest {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        // 设置接入点,在消息队列RabbitMQ版控制台实例详情页面查看。
        factory.setHost("xxx.xxx.aliyuncs.com");
        // 用户名,在消息队列RabbitMQ版控制台用户名密码管理页面查看。
        factory.setUsername("${UserName}");
        // 密码,在消息队列RabbitMQ版控制台用户名密码管理页面查看。
        factory.setPassword("${PassWord}");
        //设置为true,开启Connection自动恢复功能;设置为false,关闭Connection自动恢复功能。
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        // 设置Vhost名称,请确保已在消息队列RabbitMQ版控制台上创建完成。
        factory.setVirtualHost("${VhostName}");
        // 默认端口,非加密端口5672,加密端口5671。
        factory.setPort(5672);
        // 基于网络环境合理设置超时时间。
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);
        factory.setShutdownTimeout(0);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();        
        channel.exchangeDeclare("${ExchangeName}", "${ExchangeType}", true, false, false, null);
        channel.queueDeclare(${QueueName}", true, false, false, new HashMap<String, Object>());
        channel.queueBind(${QueueName}", ${ExchangeName}", "${BindingKey});
        // 开始发送消息。
        for (int i = 0; i < 100; i++  ) {
            // ${ExchangeName}必须在消息队列RabbitMQ版控制台上已存在,并且Exchange的类型与控制台上的类型一致。
            // BindingKey根据业务需求填入相应的BindingKey。
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(UUID.randomUUID().toString()).build();
            channel.basicPublish("${ExchangeName}", "BindingKey", true, props,
                    ("消息发送Body"  + i).getBytes(StandardCharsets.UTF_8));
        }
        connection.close();
    }
}

结果验证

您可以在函数计算控制台使用表盘解读数据指标。

  1. 登录函数计算控制台
  2. 在顶部菜单栏,选择地域。
  3. 在左侧导航栏,单击服务及函数
  4. 服务及函数页面的服务列表中,单击目标服务。
  5. 函数列表页签,找到目标函数,然后单击函数名称列下的目标函数名称。
  6. 单击日志查询页签,查看日志。
    FC Invoke Start RequestId: c2be67a7-fh1a-9619-ei4c-3c04gcf6****
    2020-11-19T11:11:34.161Z c2be67a7-fh1a-9619-ei4c-3c04gcf6c**** [verbose] Receive Event v2 ==> The event comes from aliyun.ui,event type is ui:Created:PostObject.
    2020-11-19T11:11:34.167Z c2be67a7-fh1a-9619-ei4c-3c04gcf6c**** 
    FC Invoke End RequestId: c2be67a7-fh1a-9619-ei4c-3c04gcf6c****