Producer.java

package com.dym.one2many;//發送消息import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.exception.RemotingException;public class Producer {public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {//1. 誰來發？DefaultMQProducer producer = new DefaultMQProducer("group1");//2. 發給誰？producer.setNamesrvAddr("localhost:9876");producer.start();// 3. 怎么發？// 4. 發什么？for(int i=1;i<11;i++){String msg="hello rocketmq"+i;Message message = new Message("topic2", "tag1", msg.getBytes());SendResult sendResult = producer.send(message);// 5. 發的結果是什么？System.out.println(sendResult);}//6. 打掃戰場producer.shutdown();} }

Consumer.java

package com.dym.one2many;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class Consumer {public static void main(String[] args) throws MQClientException {//1. 誰來收DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2. 從哪里收消息consumer.setNamesrvAddr("localhost:9876");//3. 監聽哪個消息隊列consumer.subscribe("topic2","*");//4. 處理業務流程 注冊監聽器consumer.registerMessageListener(new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {//寫我們的業務邏輯for(MessageExt msg:msgs){System.out.println(msg);byte[] body = msg.getBody();System.out.println(new String(body));System.out.println("==============================================");}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println("消費者起起來了");} }

?

總結

以上是生活随笔為你收集整理的RocketMQ-- 一对多 （消费者与生产者）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。