RocketMQ简单介绍
- 是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式特点。
- Producer、Consumer队列都可以分布式。
- Producer向一些队列轮流发送消息，队列集合称为 Topic，Consumer 如果做广播消费，则一个consumer实例消费这个Topic 对应的所有队列，如果做集群消费，则多个Consumer 实例平均消费这个topic对应的队列集合。（默认是集群消费）
- 能够保证严格的消息顺序（因为性能原因，不能保证消息不重复，因为总有网络不可达的情况发生，需业务端保证）。
- 提供丰富的消息拉取模式
- 高效的订阅者水平扩展能力
- 实时的消息订阅机制
- 亿级消息堆积能力
- 较少的依赖
- 支持集群部署，保证了高可用，数据不会丢失。
RocketMQ基本概念
1.Name Server：它是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
2. Broker：Broker 部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与Name Server 集群中的所有节点建立长连接，定时注册Topic信息到所有Name Server。
3. Consumer：Consumer与Name Server集群中的其中一个节点（随机选择，但不同于上一次）建立长连接，定期从Name Server 取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。
4. Producer：Producer 与Name Server集群中的其中一个节点（随机选择，但不同于上一次）建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。
RocketMQ消息传递方式
-
广播消费
一条消息被多个 Consumer 消费，即使这些 Consumer 属于同一个 Consumer Group，消息也会被 Consumer Group 中的每个 Consumer 都消费一次，广播消费中的 Consumer Group 概念可以认为在消息划分方面无意义。
在 CORBA Notification 规范中，消费方式都属于广播消费。
在 JMS 规范中，相当于 JMS publish/subscribe model
Topic发布-订阅（生产者与消费者的一对多关系）
-
集群消费
一个 Consumer Group 中的 Consumer 实例平均分摊消费消息。例如某个 Topic 有 9 条消息，其中一个 Consumer Group 有 3 个实例(可能是 3 个进程，或者 3 台机器)，那么每个实例只消费其中的 3 条消息。
在 CORBA Notification 规范中，无此消费方式。
在 JMS 规范中，JMS point-to-point model 与之类似，但是 RocketMQ 的集群消费功能大等于 PTP 模型。 因为 RocketMQ 单个 Consumer Group 内的消费者类似于 PTP，但是一个 Topic/Queue 可以被多个 Consumer Group 消费。
QUEUE点对点的方式（生产者与消费者的一对一关系）
RocketMQ 和 kafka对比
1.性能对比
Kafka单机写入TPS约在百万条/秒，消息大小10个字节。
RocketMQ单机写入TPS单实例约7万条/秒，单机部署3个Broker，可以跑到最高12万条/秒，消息大小10个字节
2.消息投递实时性
Kafka使用短轮询方式，实时性取决于轮询间隔时间
RocketMQ使用长轮询，同Push方式实时性一致，消息的投递延时通常在几个毫秒。
3.消费失败重试
Kafka消费失败不支持重试
RocketMQ消费失败支持定时重试，每次重试间隔时间顺延
4.严格的消息顺序
Kafka支持消息顺序，但是一台Broker宕机后，就会产生消息乱序
RocketMQ支持严格的消息顺序，在顺序消息场景下，一台Broker宕机后，发送消息会失败，但是不会乱序。
5.定时消息
Kafka不支持定时消息
支持定时消息
RocketMQ应用场景
异步处理
以用户注册，并且需要注册邮件和短信为例。用户注册后，需要发送注册邮件和注册短信。传统的做法有两种：串行和并行方式。如下图所示：
1）串行方式：将注册信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端。
2）并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间。
假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，并行的时间可能是100毫秒。因为CPU在单位时间内处理的请求数是一定的，假设CPU1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次（1000ms/150ms），并行方式处理的请求量是10次（1000ms/100ms）
引入消息队列后：
按照以上约定，用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了两倍。
应用解耦
以用户下单购买业务为例。
用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口。如下图
传统模式的缺点：
1）假如库存系统无法访问，则订单减库存将失败，从而导致订单失败。
2）订单系统与库存系统耦合
引入消息队列后：
1）订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。
2）库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作。
假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦。
流量削峰
流量削峰也是消息队列中的常用场景，一般在秒杀或团抢活动中使用广泛。
秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，需要在应用前端加入消息队列。
1）可以控制活动的人数。
2）可以缓解短时间内高流量压垮应用。
1）用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面。
2）秒杀业务根据消息队列中的请求信息，再做后续处理。
消息通讯
以上实际是消息队列的两种消息模式，点对点或发布订阅模式。
事务处理
比如银行转账。
如果确认消息发送失败了怎么办？RocketMQ会定期扫描消息集群中的事物消息，如果发现了Prepared消息，它会向消息发送端(生产者)确认，钱到底是减了还是没减呢？如果减了是回滚还是继续发送确认消息呢？RocketMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败
RocketMQ的不足
•消息重复问题，它不能保证不重复，只能保证正常情况下不重复
•消息过滤功能扩展比较单一
解决：消费端处理业务消息要保持幂等性，也就是同一个东西执行多次会得到相同结果，保证每条消息都有唯一编号切保证消息处理成功与去重表的日志同时出现。
入门案例
Rocketmq消费分为push和pull两种方式，push为被动消费类型，pull为主动消费类型，push方式最终还是会从broker中pull消息。不同于pull的是，push首先要注册消费监听器，当监听器处触发后才开始消费消息，所以被称为“被动”消费。通过本人实际项目的领取卡劵发送短信的需求，本案例简单呈现了消息的生产与消费的过程。实际项目可能不止一个Topic，可设置多个一组生产者对应多个Topic，每个Topic对应各自的消费者，例如本人项目卡劵的领取和卡劵的核销，可分为两个Topic生产消息，根据需求设置各自的监听器消费消息，实现业务的解耦拆分。
pom.xml设置
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.1.0-incubating</version>
</dependency>application.properties设置
aliyun.consumer.enable=rocketMQ
aliyun.producer.enable=rocketMQ
aliyun.mq.topic.chebabaCouponTopic=MQ_ORDER_COUPON
aliyun.mq.topic.chebabaCouponNotifyTopic=MQ_NOTIFY_COUPON_STATUS
aliyun.mq.producerid=PID_COUPON_001
aliyun.mq.accesss_key=xxx
aliyun.mq.secret_key=xxx
aliyun.mq.onsaddr=172.26.192.xx:9876
aliyun.mq.consumerid.coupon=CID_COUPON_001配置类初始化
@Component
public class AliwareMQConfig implements InitializingBean {
//TOPIC config
@Value("${aliyun.mq.topic.chebabaCouponTopic:}")
public String chebabaCouponTopic;
@Value("${aliyun.mq.topic.chebabaCouponNotifyTopic:}")
public String chebabaCouponNotifyTopic;
@Value("${aliyun.mq.producerid:}")
public String producerId;
@Value("${aliyun.mq.accesss_key:}")
public String accessKey;
@Value("${aliyun.mq.secret_key:}")
public String secretKey;
@Value("${aliyun.mq.onsaddr:}")
public String onsaddr;
@Value("${aliyun.mq.consumerid.coupon:}")
public String chebabaCouponConsumerId;
private Properties properties;
public Properties getProperties() {
return properties;
}
public void setProperties(Properties properties) {
this.properties = properties;
}
public String getProducerId() {
return producerId;
}
public void setProducerId(String producerId) {
this.producerId = producerId;
}
public String getAccessKey() {
return accessKey;
}
public void setAccessKey(String accessKey) {
this.accessKey = accessKey;
}
public String getSecretKey() {
return secretKey;
}
public void setSecretKey(String secretKey) {
this.secretKey = secretKey;
}
public String getOnsaddr() {
return onsaddr;
}
public void setOnsaddr(String onsaddr) {
this.onsaddr = onsaddr;
}
public String getChebabaCouponTopic() {
return chebabaCouponTopic;
}
public void setChebabaCouponTopic(String chebabaCouponTopic) {
this.chebabaCouponTopic = chebabaCouponTopic;
}
public String getChebabaCouponConsumerId() {
return chebabaCouponConsumerId;
}
public void setChebabaCouponConsumerId(String chebabaCouponConsumerId) {
this.chebabaCouponConsumerId = chebabaCouponConsumerId;
}
public String getChebabaCouponNotifyTopic() {
return chebabaCouponNotifyTopic;
}
public void setChebabaCouponNotifyTopic(String chebabaCouponNotifyTopic) {
this.chebabaCouponNotifyTopic = chebabaCouponNotifyTopic;
}
@Override
public void afterPropertiesSet() throws Exception {
Properties mqProperties = new Properties();
mqProperties.setProperty(PropertyKeyConst.ProducerId, producerId);
mqProperties.setProperty(PropertyKeyConst.AccessKey, accessKey);
mqProperties.setProperty(PropertyKeyConst.SecretKey, secretKey);
mqProperties.setProperty(PropertyKeyConst.ONSAddr, onsaddr);
this.properties = mqProperties;
}
}配置消费者和生产者
@Configuration
@ConditionalOnProperty(name = "aliyun.producer.enable", havingValue = "rocketMQ", matchIfMissing = false)
public class RocketMQConfig {
private final static Logger logger = LoggerFactory.getLogger(RocketMQConfig.class);
@Value("${aliyun.mq.onsaddr}")
private String namesrvAddr;
//System.setProperty("rocketmq.client.log.loadconfig","false");
@Bean
public Producer rocketMQProducer() throws MQClientException {
DefaultMQProducer defaultMQProducer = new DefaultMQProducer("aliPayProducer");
defaultMQProducer.setNamesrvAddr(namesrvAddr);
defaultMQProducer.start();
return new Producer() {
@Override
public boolean isStarted() {
return true;
}
@Override
public boolean isClosed() {
return false;
}
@Override
public void start() {
}
@Override
public void shutdown() {
}
@Override
public SendResult send(Message message) {
org.apache.rocketmqmon.message.Message mqMessage = new org.apache.rocketmqmon.message.Message();
mqMessage.setBody(message.getBody());
mqMessage.setTopic(message.getTopic());
mqMessage.setTags(message.getTag());
if (message.getStartDeliverTime() > 0) {
//1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
mqMessage.setDelayTimeLevel(3);//统一延迟10S处理
}
SendResult result = new SendResult();
try {
org.apache.rocketmq.client.producer.SendResult send = defaultMQProducer
.send(mqMessage);
result.setMessageId(send.getMsgId());
result.setTopic(message.getTopic());
} catch (Exception e) {
logger.error("发送本地rocketMQ消息异常", e);
result = new SendResult();
}
return result;
}
@Override
public void sendOneway(Message message) {
}
@Override
public void sendAsync(Message message, SendCallback sendCallback) {
logger.debug("[模拟发送异步MQ信息]:{}", message);
}
};
}
@Autowired
private AliwareMQConfig aliwareMQConfig;
/**
* 使用阿里云Consumer的默认配置，默认启动20个消费者进程，最大支持64个消费者
*
* @return
*/
@Bean(name = "couponConsumer")
public DefaultMQPushConsumer cluePersistenceConsumer() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(aliwareMQConfig.getChebabaCouponConsumerId());
try {
consumer.setNamesrvAddr(namesrvAddr);
consumer.subscribe(aliwareMQConfig.getChebabaCouponTopic(), "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
Message message = new Message();
message.setBody(msgs.get(0).getBody());
message.setTag(msgs.get(0).getTags());
MessageListener listener = GetSpringBeanUtil.getBean("couponMQListener",
MessageListener.class);
return RocketMQConfig.transform(listener.consume(message, null));
}
});
consumer.start();
} catch (Exception e) {
logger.error("couponConsumer init error", e);
}
logger.info("couponConsumer RocketMQ 初始化完成:{}", namesrvAddr);
return consumer;
}
public static ConsumeConcurrentlyStatus transform(Action status) {
if (Action.CommitMessage.equals(status)) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} else {
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
}
配置监听器
@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@Service
public class CouponMQListener implements MessageListener {
private final static Logger logger = LoggerFactory.getLogger(CouponMQListener.class);
@Value("${new.sms.smsServerUrl:}")
private String smsServerUrlNew;
@Autowired
private SmsService smsService;
@Override
public Action consume(Message message, ConsumeContext consumeContext) {
try{
String body = new String(message.getBody());
logger.info("卡券接收到MQ信息:{}", body);
if ("SMS".equals(message.getTag())) {
sendSms(body);
}
}catch(Exception e) {
logger.error("卡券MQ信息处理异常：{}", message, e );
return Action.ReconsumeLater;
}
return Action.CommitMessage;
}
/**
* 发送领券短信
* @param jsonString
*/
private void sendSms(String jsonString) {
JSONObject json = JSONObject.parseObject(jsonString);
Map<String, String> smsParam = new HashMap<String, String>(16);
smsParam.put("card_code",json.getString("couponCode"));
smsParam.put("card_name",json.getString("cardName"));
smsParam.put("end_date",json.getString("endDate"));
smsParam.put("name",json.getString("name"));
smsParam.put("phone",json.getString("phone"));
ResponseBean jObject = smsService.sendmess(json.getString("phone"), "0.0.0.0",
json.getString("templateCode"), "2", "领取优惠券", smsParam);
logger.info("发送卡券短信，{}|{}|{}。",jObject.getRetnCode(),jObject.getRetnDesc(),jObject.getResults());
}
}
生产消息逻辑
JSONObject json = new JSONObject();
SimpleDateFormat formatter = new SimpleDateFormat("yyyy年MM月dd日");
json.put("couponId", baseCoupon.getId());
json.put("phone", receiveCouponVO.getPhone());
json.put("name", userName);
json.put("couponCode", orderCouponDetail.getCouponCode());
json.put("cardName", baseCoupon.getCouponTitle());
json.put("endDate", formatter.format(orderCouponDetail.getExpireDate()));
//默认短信模板编码
if (StringUtil.isBlank(receiveCouponVO.getTemplateCode())) {
receiveCouponVO.setTemplateCode("SMS_107030186");
}
json.put("templateCode", receiveCouponVO.getTemplateCode());
Message message = new Message();
message.setTopic(config.getChebabaCouponTopic());
message.setStartDeliverTime(System.currentTimeMillis() + 10000); // 10s
// 查询状态
message.setTag("SMS");
message.setBody(json.toString().getBytes());
SendResult sendResult = producer.send(message);
LOGGER.info("卡券领取短信，手机号码:{}，核销码:{}。{}", receiveCouponVO.getPhone(), orderCouponDetail.getCouponCode(),
sendResult);Tip:
一个简单的案例应用，对于Rocketmq的学习不止于应用，更应于底层原理的探索。底层原理的探索也有助于提高对RocketMQ的整体理解与问题的定位，对适用场景的技术选型才更有把握。限于笔者的才疏学浅，对本文内容可能还有理解不到位的地方，如有阐述不合理之处还望留言一起探讨。
更多推荐
RocketMQ入门教程
发布评论