rocketmq初识

package com.ldj.rocketmq.producer;import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;import java.nio.charset.StandardCharsets;/*** User: ldj* Date: 2024/3/26* Time: 2:26* Description: 单向消息生产者 简单场景：如发送日志*/
public class OneWayProducer {public static void main(String[] args) {String msg = "我要和你来一场浪漫之旅！";DefaultMQProducer producer = new DefaultMQProducer("produce-group1");producer.setNamesrvAddr("192.168.208.190:9876;192.168.208.191:9876;192.168.208.192:9876");try {producer.start();producer.sendOneway(new Message("love_letter", "appointment", msg.getBytes(StandardCharsets.UTF_8)));System.out.println("消息已发送");} catch (Exception e) {e.printStackTrace();} finally {producer.shutdown();}}
}

package com.ldj.rocketmq.consumer;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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;import java.util.List;/*** User: ldj* Date: 2024/3/26* Time: 2:36* Description: 推送消费*/
public class PushConsumer {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group1");consumer.setNamesrvAddr("192.168.208.190:9876;192.168.208.191:9876;192.168.208.192:9876");//接收Topic 'love_letter'所有消息//consumer.subscribe("love_letter", "*");//只接收接收Topic 'love_letter'下的 'appointment'标签的消息consumer.subscribe("love_letter", "appointment");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.registerMessageListener(( List<MessageExt> msgs,ConsumeConcurrentlyContext context)->{msgs.forEach(msg->{System.out.println(new String(msg.getBody()));});return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});//设置消息消费模式: 集群模式（轮询：你一个我一个 || 公平：分发能者多劳）、广播模式 （每个消费者都消费全量消息）//consumer.setMessageModel(MessageModel.BROADCASTING);consumer.setMessageModel(MessageModel.CLUSTERING);consumer.start();System.out.println("消费者准备就绪");}
}