艺术网站欣赏网站注册账号
一、RocketMQ 概述
RocketMQ 是阿里巴巴开源的一款分布式消息中间件，后捐赠给 Apache 基金会成为顶级项目。它具有低延迟、高并发、高可用、高可靠等特点，广泛应用于订单交易、消息推送、流计算、日志收集等场景。
核心特点
分布式架构：支持集群部署，可水平扩展
高吞吐量：单机可支持10万级TPS
低延迟：毫秒级消息投递
高可用性：支持主从复制，自动故障转移
消息可靠性：支持消息持久化，确保不丢失
丰富的消息模式：支持普通消息、顺序消息、事务消息、定时消息等
二、核心概念
1. 基本组件
| 组件 | 说明 |
|---|---|
| NameServer | 轻量级注册中心，负责Broker的注册与发现 |
| Broker | 消息存储与转发服务器，负责消息存储、投递和查询 |
| Producer | 消息生产者，负责发送消息 |
| Consumer | 消息消费者，负责消费消息 |
| Topic | 消息主题，用于消息分类 |
| Message Queue | 消息队列，Topic的分区单位 |
| Tag | 消息标签，用于消息二级分类 |
| Group | 生产者组/消费者组，用于集群管理 |
一、阿里云rocketMQ
使用阿里云 ONS SDK
<dependency><groupId>com.aliyun.openservices</groupId><artifactId>ons-client</artifactId><version>2.0.5.Final</version> <!-- 推荐最新版本 -->
</dependency>获取阿里云 RocketMQ 配置
Endpoint：
http://{YourInstanceId}.mq-internet.aliyuncs.com:80AccessKey：阿里云账号的
AccessKey ID和AccessKey SecretTopic：消息主题（需在阿里云控制台创建）
Group ID：消费者组（需在控制台创建）
1、发消息
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;import java.util.Properties;public class AliyunMQProducer {public static void main(String[] args) {// 1. 配置 ProducerProperties properties = new Properties();properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://YourInstanceId.mq-internet.aliyuncs.com:80");properties.put(PropertyKeyConst.AccessKey, "YourAccessKey");properties.put(PropertyKeyConst.SecretKey, "YourSecretKey");properties.put(PropertyKeyConst.GROUP_ID, "YourGroupId"); // Producer Group ID// 2. 创建 ProducerProducer producer = ONSFactory.createProducer(properties);producer.start();// 3. 创建消息Message msg = new Message("YourTopic", // Topic"YourTag", // Tag"Hello Aliyun RocketMQ!".getBytes() // Body);// 4. 发送消息producer.send(msg);System.out.println("消息发送成功！");// 5. 关闭 Producerproducer.shutdown();}
}2、消费MQ
import com.aliyun.openservices.ons.api.*;
import java.util.Properties;public class AliyunMQConsumer {public static void main(String[] args) {// 1. 配置 ConsumerProperties properties = new Properties();properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://YourInstanceId.mq-internet.aliyuncs.com:80");properties.put(PropertyKeyConst.AccessKey, "YourAccessKey");properties.put(PropertyKeyConst.SecretKey, "YourSecretKey");properties.put(PropertyKeyConst.GROUP_ID, "YourGroupId"); // Consumer Group ID// 2. 创建 ConsumerConsumer consumer = ONSFactory.createConsumer(properties);// 3. 订阅 Topic 和 Tag（* 表示所有 Tag）consumer.subscribe("YourTopic", "*", new MessageListener() {@Overridepublic Action consume(Message message, ConsumeContext context) {System.out.println("收到消息: " + new String(message.getBody()));return Action.CommitMessage; // 消费成功}});// 4. 启动 Consumerconsumer.start();System.out.println("消费者已启动，等待消息...");}
}
阿里云 ONS SDK 更稳定，推荐使用（比 Apache RocketMQ 客户端更适配阿里云环境）。
Topic 和 Group ID 需先在阿里云控制台创建，否则会报错。
生产环境建议配置重试机制和日志监控，避免消息丢失。
消费模式：
集群消费（CLUSTERING）：同 Group ID 的多个 Consumer 分摊消息（默认）。
广播消费（BROADCASTING）：同 Group ID 的每个 Consumer 都收到所有消息。
二、腾讯云RocketMQ
import java.io.UnsupportedEncodingException;
import java.util.List;import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
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.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import lombok.extern.slf4j.Slf4j;/*** 腾讯云rocketMQ服务类*/
@Slf4j
@Service
@Transactional(rollbackFor = Exception.class)
public class RocketTXMqService {@Value("${rocketmq.namespace:-1}")private String namespace;@Value("${rocketmq.producer.group:-1}")private String groupName;@Value("${rocketmq.producer.access-key:-1}")private String accessKey;@Value("${rocketmq.producer.secret-key:-1}")private String secretKey;@Value("${rocketmq.name-server:-1}")private String nameserver;// MQ生产者private DefaultMQProducer producer;// MQ实例化消费者pushprivate DefaultMQPushConsumer pushConsumer;// MQ实例化消费者pullprivate DefaultLitePullConsumer pullConsumer;/*** 创建生产者* * @return*/public DefaultMQProducer getProducer() {if (null == producer) {// 实例化消息生产者Producerproducer = new DefaultMQProducer(namespace, groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL权限);// 设置NameServer的地址producer.setNamesrvAddr(nameserver);try {// 启动Producer实例producer.start();} catch (MQClientException e) {e.printStackTrace();}}return producer;}/*** 同步发送 发送消息*/public void syncSend(String topic, String tag, String data) {producer = getProducer();// 发送消息SendResult sendResult = null;try {// 创建消息实例，设置topic和消息内容Message msg = new Message(topic, tag, data.getBytes(RemotingHelper.DEFAULT_CHARSET));sendResult = producer.send(msg);log.info("埋点信息发送腾讯云MQ：" + data);log.info("发送腾讯云MQ接口返回状态sendResult:" + sendResult);} catch (UnsupportedEncodingException e) {log.error("UnsupportedEncodingException:" + e.getMessage());} catch (MQClientException e) {log.error("MQClientException:" + e.getMessage());} catch (RemotingException e) {log.error("RemotingException:" + e.getMessage());} catch (MQBrokerException e) {log.error("MQBrokerException:" + e.getMessage());} catch (InterruptedException e) {log.error("InterruptedException:" + e.getMessage());}}/*** 创建push消费者* * @return*/public DefaultMQPushConsumer getPushConsumer() {if (null == pushConsumer) {// 实例化消费者pushConsumer = new DefaultMQPushConsumer(namespace, groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); // ACL权限// 设置NameServer的地址pushConsumer.setNamesrvAddr(nameserver);}return pushConsumer;}/*** 创建pull 消费者* * @return*/public DefaultLitePullConsumer getPullConsumer() {if (null == pullConsumer) {// 实例化消费者// 实例化消费者pullConsumer = new DefaultLitePullConsumer(namespace, groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));// 设置NameServer的地址pullConsumer.setNamesrvAddr(nameserver);// 设置从第一个偏移量开始消费pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);}return pullConsumer;}/*** push方式订阅消费* * @param topicName*/public void pushConsumer(String topicName) {pushConsumer = this.getPushConsumer();if (null != pushConsumer) {try {pushConsumer.subscribe(topicName, "*");// 注册回调实现类来处理从broker拉取回来的消息pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {// 消息处理逻辑log.info("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);// 标记该消息已经被成功消费, 根据消费情况，返回处理状态return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 启动消费者实例pushConsumer.start();} catch (MQClientException e) {log.error("push MQClientException:" + e.getMessage());}}}/*** pull方式订阅消费* * @param topicName*/public void pullConsumer(String topicName) {pullConsumer = this.getPullConsumer();if (null != pullConsumer) {try {// 订阅topicpullConsumer.subscribe(topicName, "*");// 启动消费者实例pullConsumer.start();} catch (MQClientException e) {log.error(" pull MQClientException:" + e.getMessage());}try {log.info("Consumer Started.%n");while (true) {// 拉取消息List<MessageExt> messageExts = pullConsumer.poll();log.info("%s%n", messageExts);}} finally {pullConsumer.shutdown();}}}}
