SpringBoot整合RocketMQ

1. <dependency>
2.     <groupId>org.apache.rocketmq</groupId>
3.     <artifactId>rocketmq-spring-boot-starter</artifactId>
4.     <version>2.0.3</version>
5. </dependency>

1. @RestController
2. @RequestMapping("/demo")
3. public class SendController {
4. 
5. @Autowired
6.     RocketMQTemplate rocketMQTemplate;//模板类：建连接及断连结
7. 
8. @GetMapping("/send")
9. public String send(){
10. 
11. User smoky = new User("smoky", 18);
12. 
13. //同步消息
14. SendResult syncSend = rocketMQTemplate.syncSend("topic10", smoky, 3);
15. 
16. return "success!";
17. }

1. //异步消息
2.         rocketMQTemplate.asyncSend("topic10", smoky, new SendCallback() {
3. @Override
4. public void onSuccess(SendResult sendResult) {
5. //成功回调函数
6.                 System.out.println("sendResult = " + sendResult);
7.             }
8. 
9. @Override
10. public void onException(Throwable throwable) {
11. //发送失败回调函数
12.                 System.out.println(throwable);
13.             }
14.         },1000);

rocketMQTemplate.sendOneWay("topic10",smoky);

1. //延时消息
2.         rocketMQTemplate.syncSend("topic10", MessageBuilder.withPayload(msg).build(),2000,2);

1. @Service
2. //根据sql过滤
3. @RocketMQMessageListener(topic = "topic10",consumerGroup = "group1",
4.         selectorType = SelectorType.SQL92,
5.         selectorExpression = "age>92",
6.         messageModel = MessageModel.BROADCASTING //广播模式
7.        )
8. 
9. //sql过滤
10. public class DemoConsumer implements RocketMQListener<User> {
11. 
12. @Override
13. public void onMessage(User user) {
14. //业务逻辑
15.         System.out.println("user = " + user);
16.     }
17. }