RocketMq

RocketMq用法

有三个对象
Producer，Consumer，Topic
简单来说就是Producer发送给Topic
然后消费者拿去消费

三者都需要在注册中心注册

整合

导包

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependence>

生产者代码

public static void main(String[] args){
// 新增消息⽣生产者
DefaultMQProucer producer = new DefaultMQProucer("producer_group");
// 配置注册中⼼心
producer.setNamesrvAddr("localhost:9876");
// 启动
producer.start();
// 新建消息对象
Message message = new
Message("topicA","message".context.getBytes(Charset.forName("utf-8")));
// 发送消息
producer.send(message);
}
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer mqConsumer = new
DefaultMQPushConsumer("consumer_group");
mqConsumer.setNamesrvAddr("localhost:9876");
mqConsumer.subscribe("topicA", "*");
// 设置消息监听器器
mqConsumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
msgs, ConsumeConcurrentlyContext context) {
MessageExt message = msgs.get(0);
//获取消息内容
byte[] body = message.getBody();
});

消费者代码

public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer mqConsumer = new
DefaultMQPushConsumer("consumer_group");
mqConsumer.setNamesrvAddr("localhost:9876");
mqConsumer.subscribe("topicA", "*");
// 设置消息监听器器
mqConsumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
msgs, ConsumeConcurrentlyContext context) {
MessageExt message = msgs.get(0);
//获取消息内容
byte[] body = message.getBody();
});