pom文件

  1. <dependency>
  2.   <groupId>com.aliyun.openservices</groupId>
  3.   <artifactId>ons-client</artifactId>
  4.   <version>1.8.4.Final</version>
  5. </dependency>

配置文件

  1. 配置一些基础的配置信息
  1. #启动测试之前请替换如下 XXX 为您的配置
  2. rocketmq.accessKey=XXX
  3. rocketmq.secretKey=XX
  4. rocketmq.nameSrvAddr=XXX
  6. rocketmq.topic=XXX
  7. rocketmq.groupId=XXX
  8. rocketmq.tag=*
  10. rocketmq.orderTopic=XXX
  11. rocketmq.orderGroupId=XXX
  12. rocketmq.orderTag=*
  1. 创建一个配置项实体类，用于读取配置信息
  1. @Configuration
  2. @ConfigurationProperties(prefix = "rocketmq")
  3. @Date
  4. public class MqConfig {
  6.     private String accessKey;
  7.     private String secretKey;
  8.     private String nameSrvAddr;
  9.     private String topic;
  10.     private String groupId;
  11.     private String tag;
  12.     private String orderTopic;
  13.     private String orderGroupId;
  14.     private String orderTag;
  16.     public Properties getMqPropertie() {
  17.         Properties properties = new Properties();
  18.         properties.setProperty(PropertyKeyConst.AccessKey, this.accessKey);
  19.         properties.setProperty(PropertyKeyConst.SecretKey, this.secretKey);
  20.         properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);
  21.         return properties;
  22.     }
  23. }

普通类型的生产者和消费者定义

生产者

建立连接

  1. @Configuration
  2. public class ProducerClient {
  4.     @Autowired
  5.     private MqConfig mqConfig;
  7.     @Bean(initMethod = "start", destroyMethod = "shutdown")
  8.     public ProducerBean buildProducer() {
  9.         ProducerBean producer = new ProducerBean();
  10.         producer.setProperties(mqConfig.getMqPropertie());
  11.         return producer;
  12.     }
  14. }

生产者发送消息

同步

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest
  3. public class SyncProducerTest {
  5.     //普通消息的Producer 已经注册到了spring容器中，后面需要使用时可以直接注入到其它类中
  6.     @Autowired
  7.     private ProducerBean producer;
  9.     @Autowired
  10.     private MqConfig mqConfig;
  12.     @Test
  13.     public void testSend() {
  14.         //循环发送消息
  15.         for (int i = 0; i < 100; i++) {
  16.             Message msg = new Message( //
  17.                     // Message所属的Topic
  18.                     mqConfig.getTopic(),
  19.                     // Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
  20.                     mqConfig.getTag(),
  21.                     // Message Body 可以是任何二进制形式的数据， MQ不做任何干预
  22.                     // 需要Producer与Consumer协商好一致的序列化和反序列化方式
  23.                     "Hello MQ".getBytes());
  24.             // 设置代表消息的业务关键属性，请尽可能全局唯一
  25.             // 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发
  26.             // 注意：不设置也不会影响消息正常收发
  27.             msg.setKey("ORDERID_100");
  28.             // 发送消息，只要不抛异常就是成功
  29.             try {
  30.                 SendResult sendResult = producer.send(msg);
  31.                 assert sendResult != null;
  32.                 System.out.println(sendResult);
  33.             } catch (ONSClientException e) {
  34.                 System.out.println("发送失败");
  35.                 //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
  36.             }
  37.         }
  38.     }
  41. }

异步

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest
  3. public class AsyncProducerTest {
  5.     //普通消息的Producer 已经注册到了spring容器中，后面需要使用时可以直接注入到其它类中
  6.     @Autowired
  7.     private ProducerBean producer;
  9.     @Autowired
  10.     private MqConfig mqConfig;
  12.     @Test
  13.     public void testSend() {
  15.         //对于使用异步接口，建议设置单独的回调处理线程池，拥有更灵活的配置和监控能力。
  16.         //如下构造线程的方式请求队列为无界仅用作示例，有OOM的风险。
  17.         //更合理的构造方式请参考阿里巴巴Java开发手册：https://github.com/alibaba/p3c
  18.         producer.setCallbackExecutor(Executors.newFixedThreadPool(10));
  20.         //循环发送消息
  21.         for (int i = 0; i < 1; i++) {
  22.             Message msg = new Message( //
  23.                     // Message所属的Topic
  24.                     mqConfig.getTopic(),
  25.                     // Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
  26.                     mqConfig.getTag(),
  27.                     // Message Body 可以是任何二进制形式的数据， MQ不做任何干预
  28.                     // 需要Producer与Consumer协商好一致的序列化和反序列化方式
  29.                     "Hello MQ".getBytes());
  30.             // 设置代表消息的业务关键属性，请尽可能全局唯一
  31.             // 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发
  32.             // 注意：不设置也不会影响消息正常收发
  33.             msg.setKey("ORDERID_100");
  34.             // 发送消息，只要不抛异常就是成功
  35.             try {
  36.                 producer.sendAsync(msg, new SendCallback() {
  37.                     @Override
  38.                     public void onSuccess(final SendResult sendResult) {
  39.                         assert sendResult != null;
  40.                         System.out.println(sendResult);
  41.                     }
  43.                     @Override
  44.                     public void onException(final OnExceptionContext context) {
  45.                         //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
  46.                     }
  47.                 });
  48.             } catch (ONSClientException e) {
  49.                 System.out.println("发送失败");
  50.                 //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
  51.             }
  52.         }
  53.     }
  55. }

sql 过滤

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest
  3. public class SqlProducerTest {
  5.     //普通消息的Producer 已经注册到了spring容器中，后面需要使用时可以直接注入到其它类中
  6.     @Autowired
  7.     private ProducerBean producer;
  9.     @Autowired
  10.     private MqConfig mqConfig;
  12.     @Test
  13.     public void testSend() {
  15.         //循环发送消息
  16.         for (int i = 0; i < 100; i++) {
  17.             String tag;
  18.             int div = i % 3;
  19.             if (div == 0) {
  20.                 tag = "TagA";
  21.             } else if (div == 1) {
  22.                 tag = "TagB";
  23.             } else {
  24.                 tag = "TagC";
  25.             }
  27.             Message msg = new Message( //
  28.                     // Message所属的Topic
  29.                     mqConfig.getTopic(),
  30.                     // Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
  31.                     tag,
  32.                     // Message Body 可以是任何二进制形式的数据， MQ不做任何干预
  33.                     // 需要Producer与Consumer协商好一致的序列化和反序列化方式
  34.                     "Hello MQ".getBytes());
  35.             // 设置代表消息的业务关键属性，请尽可能全局唯一
  36.             // 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发
  37.             // 注意：不设置也不会影响消息正常收发
  38.             msg.setKey("ORDERID_100");
  40.             // 设置自定义属性，该属性可用于做SQL属性过滤
  41.             msg.putUserProperties("a", String.valueOf(i));
  43.             // 发送消息，只要不抛异常就是成功
  44.             try {
  45.                 SendResult sendResult = producer.send(msg);
  46.                 assert sendResult != null;
  47.                 System.out.println(sendResult);
  48.             } catch (ONSClientException e) {
  49.                 System.out.println("发送失败");
  50.                 //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
  51.             }
  52.         }
  54.     }
  56. }

消费者

监听

  1. @Component
  2. public class DemoMessageListener implements MessageListener {
  4.     @Override
  5.     public Action consume(Message message, ConsumeContext context) {
  7.         System.out.println("Receive: " + message);
  8.         try {
  9.             //do something..
  10.             return Action.CommitMessage;
  11.         } catch (Exception e) {
  12.             //消费失败
  13.             return Action.ReconsumeLater;
  14.         }
  15.     }
  16. }

普通消费方式

  1. //项目中加上 @Configuration 注解，这样服务启动时consumer也启动了
  2. public class ConsumerClient {
  4.     @Autowired
  5.     private MqConfig mqConfig;
  7.     @Autowired
  8.     private DemoMessageListener messageListener;
  10.     @Bean(initMethod = "start", destroyMethod = "shutdown")
  11.     public ConsumerBean buildConsumer() {
  12.         ConsumerBean consumerBean = new ConsumerBean();
  13.         //配置文件
  14.         Properties properties = mqConfig.getMqPropertie();
  15.         properties.setProperty(PropertyKeyConst.GROUP_ID, mqConfig.getGroupId());
  16.         //将消费者线程数固定为20个 20为默认值
  17.         properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
  18.         consumerBean.setProperties(properties);
  19.         //订阅关系
  20.         Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
  21.         Subscription subscription = new Subscription();
  22.         subscription.setTopic(mqConfig.getTopic());
  23.         subscription.setExpression(mqConfig.getTag());
  24.         subscriptionTable.put(subscription, messageListener);
  25.         //订阅多个topic如上面设置
  27.         consumerBean.setSubscriptionTable(subscriptionTable);
  28.         return consumerBean;
  29.     }
  31. }

sql 过滤消费

  1. //正式开发时可以加上 @Configuration 注解，这样服务启动时consumer也启动了
  2. //sql92只有mq铂金版才支持
  3. public class SqlConsumerClient {
  5.     @Autowired
  6.     private MqConfig mqConfig;
  8.     @Autowired
  9.     private DemoMessageListener messageListener;
  11.     @Bean(initMethod = "start", destroyMethod = "shutdown")
  12.     public ConsumerBean buildSqlConsumer() {
  13.         ConsumerBean consumerBean = new ConsumerBean();
  14.         //配置文件
  15.         Properties properties = mqConfig.getMqPropertie();
  16.         properties.setProperty(PropertyKeyConst.GROUP_ID, mqConfig.getGroupId());
  17.         consumerBean.setProperties(properties);
  18.         //订阅关系
  19.         Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
  20.         Subscription subscription = new Subscription();
  21.         subscription.setTopic(mqConfig.getTopic());
  22.         // 表示需要使用SQL来过滤消息
  23.         subscription.setType("SQL92");
  24.         //需要消息的tag是'TagA'或'TagB'并且自定义属性a(在发送消息的时候通过putUserProperties方法放入)需要在[0,3]
  25.         //SQL过滤同样可以使用消息的tag作为过滤条件(消息的tag在消息的属性中叫做 TAGS)
  26.         //SQL过滤同样可以在顺序消费中使用
  27.         subscription.setExpression("(TAGS is not null and TAGS in ('TagA', 'TagB')) and (a is not null and a between 0 and 3)");
  28.         subscriptionTable.put(subscription, messageListener);
  29.         //订阅多个topic如上面设置
  31.         consumerBean.setSubscriptionTable(subscriptionTable);
  32.         return consumerBean;
  33.     }
  35. }