pom.xml

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>dym-rocketmq-test</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.8.0</version></dependency></dependencies> </project>

Producer.java

package com.dym.simple;//發送消息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. 發什么？String msg="hello rocketmq";Message message = new Message("topic1", "tag1", msg.getBytes());SendResult sendResult = producer.send(message);// 5. 發的結果是什么？System.out.println(sendResult);//6. 打掃戰場producer.shutdown();} }

Consumer.java

package com.dym.simple;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.client.producer.DefaultMQProducer; 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("topic1","*");//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));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println("消費者起起來了");} }

?

總結

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

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