目录
环境
- JDK11、SpringBoot2.6.3、RocketMQ4.8、IDEA
生产者
application.yml
server:
port: 6101
spring:
application:
name: lik-rocketmq-producer
rocketmq:
name-server: 127.0.0.1:9876
producer:
group: mq-producer-1
发送普通消息
package com.lik;
import cn.hutool.core.util.IdUtil;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class ProducerTest {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Test
public void demo1() {
for (int i = 0; i < 10; i++) {
rocketMQTemplate.syncSend("topic-A", "消息 " + i + " " + IdUtil.getSnowflake(1, 1).nextId());
}
}
}
消费者
application.yml
server:
port: 6103
spring:
application:
name: lik-rocketmq-consumer
rocketmq:
name-server: 127.0.0.1:9876
消费组（consumerGroup）集群
当consumerGroup 消费组有多个，集群消费同一个主题时，默认情况下这个主题的消息会分流到这个消费组的不同消费者进行消费，不同消费组之间可以同时消费同一个消息
消费者1 consumer-group-1 （集群）
package com.lik.listener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup="consumer-group-1", topic="topic-A")
public class TestConsumer1 implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("消费者1: " + message);
}
}
消费者2 consumer-group-1 （集群）
package com.lik.listener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup="consumer-group-1", topic="topic-A")
public class TestConsumer2 implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("消费者2: " + message);
}
}
消费者3 consumer-group-2
package com.lik.listener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup="consumer-group-2", topic="topic-A")
public class TestConsumer3 implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("消费者3: " + message);
}
}
消息消费结果
集群模式小结
消费者1 和 消费者2同为一个消费组并订阅同一个主题时，默认情况下会平分这个主题的消息进行消费。
消费者3 跟 消费者1、2不在同一个消费组，会单独消费这个主题下的所有消息。
广播模式消费
默认情况下同一消费组群集时会分滩来消费一个主题的消息。
如果特殊情况需要集群下的所有消费者都消费这个主题的消息时，可将消费模式设置为广播模式 messageModel = MessageModel.BROADCASTING
消费者5
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "consumer-group-3", topic = "topic-B", messageModel = MessageModel.BROADCASTING)
public class TestConsumer5 implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("消费者5: " + message);
}
}
消费者6
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "consumer-group-3", topic = "topic-B", messageModel = MessageModel.BROADCASTING)
public class TestConsumer6 implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("消费者6: " + message);
}
}
广播模式消费结果
广播模式小结
消费者5 和 消费者6都同为一个消费组（consumer-group-3），并订阅同一个主题（topic-B），只需要在@RocketMQMessageListener注解中添加 messageModel = MessageModel.BROADCASTING 参数设置为广播模式，即可实现每个消费者都消费这个主题的所有消息。