Java SDK如何订阅消息

本文介绍如何通过 MQ SDK 进行消息订阅。
请确保同一个 Consumer ID 下所有 Consumer 实例的订阅关系保持一致，具体请参考 订阅关系一致文档。
TCP 接入点域名，请 前往查看。
MQ 支持两种订阅方式。

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

1. import com.aliyun.openservices.ons.api.Action;
2. import com.aliyun.openservices.ons.api.ConsumeContext;
3. import com.aliyun.openservices.ons.api.Consumer;
4. import com.aliyun.openservices.ons.api.Message;
5. import com.aliyun.openservices.ons.api.MessageListener;
6. import com.aliyun.openservices.ons.api.ONSFactory;
7. import com.aliyun.openservices.ons.api.PropertyKeyConst;
9. import java.util.Properties;
11. public class ConsumerTest {
12.     public static void main(String[] args) {
13.         Properties properties = new Properties();
14.         // 您在控制台创建的 Consumer ID
15.         properties.put(PropertyKeyConst.ConsumerId, "XXX");
16.         // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
17.         properties.put(PropertyKeyConst.AccessKey, "XXX");
18.         // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
19.         properties.put(PropertyKeyConst.SecretKey, "XXX");
20.         // 设置 TCP 接入域名（此处以公共云生产环境为例）
21.         properties.put(PropertyKeyConst.ONSAddr,
22.           "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal");
23.           // 集群订阅方式 (默认)
24.           // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
25.           // 广播订阅方式
26.           // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
28.         Consumer consumer = ONSFactory.createConsumer(properties);
29.         consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() { //订阅多个Tag
30.             public Action consume(Message message, ConsumeContext context) {
31.                 System.out.println("Receive: " + message);
32.                 return Action.CommitMessage;
33.             }
34.         });
36.         //订阅另外一个Topic
37.         consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() { //订阅全部Tag
38.             public Action consume(Message message, ConsumeContext context) {
39.                 System.out.println("Receive: " + message);
40.                 return Action.CommitMessage;
41.             }
42.         });
44.         consumer.start();
45.         System.out.println("Consumer Started");
46.     }
47. }