Alibaba阿里 RocketMQ入门实例(订阅消息)

比眉伴天荒 2022-02-25 14:28 619阅读 0赞

## 订阅方式 ##

消息队列 RocketMQ 支持以下两种订阅方式：

*  **集群订阅：**同一个 Group ID 所标识的所有 Consumer 平均分摊消费消息。 例如某个 Topic 有 9 条消息，一个 Group ID 有 3 个 Consumer 实例，那么在集群消费模式下每个实例平均分摊，只消费其中的 3 条消息。
    
        // 集群订阅方式设置（不设置的情况下，默认为集群订阅方式）
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
 *  **广播订阅：**同一个 Group ID 所标识的所有 Consumer 都会各自消费某条消息一次。 例如某个 Topic 有 9 条消息，一个 Group ID 有 3 个 Consumer 实例，那么在广播消费模式下每个实例都会各自消费 9 条消息。
    
        // 广播订阅方式设置
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

## 示例代码 ##

1.      import com.aliyun.openservices.ons.api.Action;
        import com.aliyun.openservices.ons.api.ConsumeContext;
        import com.aliyun.openservices.ons.api.Consumer;
        import com.aliyun.openservices.ons.api.Message;
        import com.aliyun.openservices.ons.api.MessageListener;
        import com.aliyun.openservices.ons.api.ONSFactory;
        import com.aliyun.openservices.ons.api.PropertyKeyConst;
        
        import java.util.Properties;
        
        public class ConsumerTest {
        public static void main(String[] args) {
        Properties properties = new Properties();
        // 您在控制台创建的 Group ID
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.AccessKey, "XXX");
        // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
        properties.put(PropertyKeyConst.NAMESRV_ADDR,
        "XXX");
        // 集群订阅方式 (默认)
        // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        // 广播订阅方式
        // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() { //订阅多个 Tag
        public Action consume(Message message, ConsumeContext context) {
        System.out.println("Receive: " + message);
        return Action.CommitMessage;
        }
        });
        
        //订阅另外一个 Topic
        consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() { //订阅全部 Tag
        public Action consume(Message message, ConsumeContext context) {
        System.out.println("Receive: " + message);
        return Action.CommitMessage;
        }
        });
        
        consumer.start();
        System.out.println("Consumer Started");
        }
        }

**注意：**

*  在广播消费模式下，您无法在控制台设置消息堆积报警，也无法进行消息堆积查询。 因此，您也可以创建多个 Group ID 来达到广播模式的效果，请参见[集群消费和广播消费][Link 1]中**使用集群模式模拟广播**一节。

[Link 1]: https://help.aliyun.com/document_detail/43163.html