RocketMQ的demo代码
下面是一个使用Java实现的RocketMQ示例代码，用于发送和消费消息：
首先，您需要下载并安装RocketMQ，并启动NameServer和Broker。
接下来，您可以使用以下示例代码来发送和消费消息：
Producer.java文件：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;public class Producer {public static void main(String[] args) {try {// 创建一个DefaultMQProducer实例DefaultMQProducer producer = new DefaultMQProducer("producer_group");// 设置NameServer地址producer.setNamesrvAddr("localhost:9876");// 启动Producer实例producer.start();// 创建消息对象Message message = new Message("topic_name", "tag", "Hello, RocketMQ!".getBytes());// 发送消息producer.send(message);// 关闭Producer实例producer.shutdown();} catch (Exception e) {e.printStackTrace();}}
}
Consumer.java文件：
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;public class Consumer {public static void main(String[] args) {try {// 创建一个DefaultMQPushConsumer实例DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");// 设置NameServer地址consumer.setNamesrvAddr("localhost:9876");// 设置消费开始位置consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);// 订阅主题和标签consumer.subscribe("topic_name", "tag");// 注册消息监听器consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {for (MessageExt msg : msgs) {System.out.println("Received message: " + new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 启动Consumer实例consumer.start();} catch (Exception e) {e.printStackTrace();}}
}
在上述示例代码中，Producer类用于发送消息，而Consumer类用于接收和处理消息。
在Producer类中，我们创建一个DefaultMQProducer实例，并设置NameServer的地址。然后，我们创建一个消息对象，指定主题、标签和消息内容。最后，通过调用send方法发送消息。
在Consumer类中，我们创建一个DefaultMQPushConsumer实例，并设置NameServer的地址。然后，我们设置消费开始位置和订阅特定的主题和标签。注册一个消息监听器，用于处理接收到的消息。最后，调用start方法启动Consumer实例。
请确保您已正确配置RocketMQ服务器和相关依赖项，并根据需要更改服务器地址、主题和标签。运行Producer和Consumer代码后，您将看到Producer发送的消息在Consumer端被接收和打印出来。
当然！这里是另一个使用Java实现的RocketMQ示例代码，用于发送事务消息：
首先，您需要下载并安装RocketMQ，并启动NameServer和Broker。
接下来，您可以使用以下示例代码来发送和处理事务消息：
TransactionProducer.java文件：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;public class TransactionProducer {public static void main(String[] args) {try {// 创建一个TransactionMQProducer实例TransactionMQProducer producer = new TransactionMQProducer("producer_group");// 设置NameServer地址producer.setNamesrvAddr("localhost:9876");// 设置事务监听器producer.setTransactionListener(new TransactionListener() {@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {// 执行本地事务，根据业务逻辑返回LocalTransactionState// 如果事务执行成功，返回COMMIT_MESSAGE// 如果事务执行失败，返回ROLLBACK_MESSAGE// 如果事务状态不确定，返回UNKNOWreturn LocalTransactionState.COMMIT_MESSAGE;}@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {// 检查本地事务状态，根据消息内容返回LocalTransactionState// 如果事务已提交，返回COMMIT_MESSAGE// 如果事务已回滚，返回ROLLBACK_MESSAGE// 如果事务状态不确定，返回UNKNOWreturn LocalTransactionState.COMMIT_MESSAGE;}});// 启动Producer实例producer.start();// 创建消息对象Message message = new Message("topic_name", "tag", "Hello, RocketMQ!".getBytes());// 发送事务消息TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);System.out.println("Transaction Send Result: " + sendResult.getLocalTransactionState());// 关闭Producer实例producer.shutdown();} catch (Exception e) {e.printStackTrace();}}
}
在上述示例代码中，我们创建了一个TransactionMQProducer实例，并设置了NameServer的地址。然后，我们设置了事务监听器，其中包含executeLocalTransaction方法和checkLocalTransaction方法。在executeLocalTransaction方法中，您可以执行本地事务操作，并根据事务结果返回适当的LocalTransactionState。在checkLocalTransaction方法中，您可以检查本地事务状态，并返回相应的LocalTransactionState。
在主程序中，我们创建了一个消息对象，并通过调用sendMessageInTransaction方法发送事务消息。然后，我们打印事务发送结果的LocalTransactionState。
请确保您已正确配置RocketMQ服务器和相关依赖项，并根据需要更改服务器地址、主题和标签。运行TransactionProducer代码后，您将看到事务消息发送的结果。根据本地事务执行的结果和检查的结果，LocalTransactionState将被设置为相应的状态（
COMMIT_MESSAGE、ROLLBACK_MESSAGE或UNKNOW）。
当然！这里是另一个使用Java实现的RocketMQ示例代码，用于顺序消息的发送和消费：
首先，您需要下载并安装RocketMQ，并启动NameServer和Broker。
接下来，您可以使用以下示例代码来发送和消费顺序消息：
顺序消息生产者 (OrderedProducer.java)：
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.common.message.MessageQueueSelector;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;import java.util.List;public class OrderedProducer {public static void main(String[] args) {try {// 创建一个DefaultMQProducer实例DefaultMQProducer producer = new DefaultMQProducer("producer_group");// 设置NameServer地址producer.setNamesrvAddr("localhost:9876");// 启动Producer实例producer.start();// 发送顺序消息for (int i = 0; i < 10; i++) {Message message = new Message("topic_name", "tag", ("Hello, RocketMQ! " + i).getBytes());SendResult sendResult = producer.send(message, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Integer orderId = (Integer) arg;int index = orderId % mqs.size();return mqs.get(index);}}, i);System.out.println("Send Result: " + sendResult);}// 关闭Producer实例producer.shutdown();} catch (Exception e) {e.printStackTrace();}}
}
顺序消息消费者 (OrderedConsumer.java)：
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.common.message.MessageExt;import java.util.List;public class OrderedConsumer {public static void main(String[] args) {try {// 创建一个DefaultMQPushConsumer实例DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");// 设置NameServer地址consumer.setNamesrvAddr("localhost:9876");// 设置消费模式为集群模式consumer.setMessageModel(MessageModel.CLUSTERING);// 订阅主题和标签consumer.subscribe("topic_name", "tag");// 注册消息监听器consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {for (MessageExt msg : msgs) {System.out.println("Received message: " + new String(msg.getBody()) + ", QueueId: " + msg.getQueueId());}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 启动Consumer实例consumer.start();} catch (Exception e) {e.printStackTrace();}}
}
在上述示例代码中，OrderedProducer类用于发送顺序消息，而OrderedConsumer类用于接收和处理顺序消息。
在OrderedProducer类中，我们创建了一个DefaultMQProducer实例，并设置了Name
Server的地址。然后，我们使用循环发送了10条顺序消息。在发送消息时，我们使用MessageQueueSelector来选择要发送的消息队列，并通过参数指定了消息的顺序。每条消息都会返回一个SendResult对象。
在OrderedConsumer类中，我们创建了一个DefaultMQPushConsumer实例，并设置了NameServer的地址。我们还设置了消费模式为集群模式，通过subscribe方法订阅特定的主题和标签。注册了一个消息监听器，用于处理接收到的消息。收到的消息将被打印出来，并附带消息所在的队列ID。
请确保您已正确配置RocketMQ服务器和相关依赖项，并根据需要更改服务器地址、主题和标签。运行OrderedProducer和OrderedConsumer代码后，您将看到顺序消息发送和消费的结果。每条消息都将按照发送时指定的顺序被消费。