青岛中小微企业互联网站建设补贴/网站开发工具

环境

jdk1.8, springboot2.7.3

Maven依赖

        <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.3</version><relativePath/> <!-- lookup parent from repository --></parent>        <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.2</version><exclusions><exclusion><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.5</version></dependency>

配置文件

rocketmq.name-server=192.168.6.128:9876
#生产通用群组, 也可单独指定
rocketmq.producer.group=springBootGroup
#消费通用群组, 也可单独指定
rocketmq.consumer.group=testGroup
server.port=9000

代码

生产者发送消息

@RestController
@RequestMapping("/producer")
public class ProducerController {@Autowiredprivate ProducerService producerService;// 发送同步消息@PostMapping("/sendSync")public Object sendSync(@RequestBody MessageReq req) {return producerService.sendSyncMessage(req.getTopic(), req.getTag(), req.getMessage());}// 发送异步消息@PostMapping("/sendAsync")public Object sendAsyncMessage(@RequestBody MessageReq req) {producerService.sendAsyncMessage(req.getTopic(), req.getTag(), req.getMessage());return "200";}
}

@Service
public class ProducerService {@Autowiredprivate RocketMQTemplate rocketMQTemplate;/*** 发送同步消息.* @return 发送结果*/public SendResult sendSyncMessage(String topic, String tag, String message) {// param1: topic和tag冒号分隔return rocketMQTemplate.syncSend(topic + ":" + tag, message);}/*** 发送异步消息.*/public void sendAsyncMessage(String topic, String tag, String message) {rocketMQTemplate.convertAndSend(topic + ":" + tag, message);}
}

消费者

@Component
@RocketMQMessageListener(consumerGroup = "SimpleStringConsumerGroup",  // consumerGroup：消费者组名topic = "MQ_sp_test1",                         // topic：订阅的主题selectorExpression = "Tag-kk||Tag-kk2",         // selectorExpression, 1. 根据Tag过滤, 多个用||分割, 也可设置*; 2. 根据SQL92语法过滤
//        selectorExpression = "*",
//        selectorType = SelectorType.SQL92,             // 设置SQL92语法过滤, 不设置默认TAGmessageModel = MessageModel.CLUSTERING,  // messageModel: 控制消息模式。MessageModel.CLUSTERING：负载均衡；MessageModel.BROADCASTING：广播模式consumeMode= ConsumeMode.CONCURRENTLY    // CONCURRENTLY: 无序消费; ORDERLY: 有序消费
)
public class SimpleConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received message : "+ message);}
}

测试

同步消息

异步消息 

TAG过滤消息

1. 消费者指定了TAG, 不满足的不会消费, 状态是CONSUMED_BUT_FILTERED

消费端接收消息