1.nameserver是无状态的,控制着broker,producer,consumer集群的同步
2.broker是处理消息中转,负责储存和过滤消息
3.rocketmq有两种消息消费模型,广播模式,一个消费者组中的每一个消费者都消费;集群模式,一个消费者组中的每一个消费者平均消费消息
4.顺序消费,生产者向一个topic的一个队列,顺序发送消息,切生产者只能单线程发送,消费者在消费时使用顺序消费模式
rocketmq是什么
1.一种队列模型的消息中间件,高性能,高可靠,高实时,分布式
2.producer,consumer,broker都可以分布式
3.默认情况下,prducer会向一个topic下的一些队列轮流发送消息
api
生产者
//顺序发送 DefaultMQProducer producer = new DefaultMQProducer(PRODUCT_GROUP); producer.setNamesrvAddr(ROCKET_SERVER); Message message = new Message("test", "tags", "1", info.getBytes()); producer.start(); for (int i = 0; i < 10; i++) { SendResult result = producer.send(message,new SelectMessageQueueByHash(),i); } producer.shutdown();
//随机发送 DefaultMQProducer producer = new DefaultMQProducer(PRODUCT_GROUP); producer.setNamesrvAddr(ROCKET_SERVER); Message message = new Message("test", "tags", "1", info.getBytes()); producer.start(); for (int i = 0; i < 10; i++) { SendResult result = producer.send(message); } producer.shutdown();
消费者
//普通消费 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); consumer.setNamesrvAddr(ROCKET_SERVER); consumer.subscribe("test", "*"); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { if (CollectionUtils.isNotEmpty(list)) { Iterator it = list.iterator(); while (it.hasNext()) { MessageExt msg = (MessageExt) it.next(); System.out.println(msg.toString()); } } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start();
//顺序消费 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); consumer.setNamesrvAddr(ROCKET_SERVER); consumer.subscribe("test", "*"); consumer.registerMessageListener(new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) { if (CollectionUtils.isNotEmpty(list)) { Iterator it = list.iterator(); while (it.hasNext()) { MessageExt msg = (MessageExt) it.next(); System.out.println(msg.toString()); } } return ConsumeOrderlyStatus.SUCCESS; } }); consumer.start();
启动广播模式
consumer.setMessageModel(MessageModel.BROADCASTING);
相关推荐
安装 RocketMQ 1. 下载 rocketmq 二进制文件 wget --no-check-certificate https://dlcdn.apache.org/rocketmq/4.9.2/rocketmq-all-4.9.2-bin-release.zip 2. 解压缩 rocketmq 将 rocketmq 解压到 /usr/local/ 目录 ...
赠送jar包:rocketmq-client-3.6.2.Final.jar; 赠送原API文档:rocketmq-client-3.6.2.Final-javadoc.jar; 赠送源代码:rocketmq-client-3.6.2.Final-sources.jar; 包含翻译后的API文档:rocketmq-client-...
cd docker-rocketmq cd rmq chmod +x start.sh ./start.sh 不能设置权限777的同学可以设置如下 chown 3000:3000 ./rmqs/logs chown 3000:3000 ./rmqs/store chown 3000:3000 ./rmq/logs chown 3000:3000 ./rmq/...
1 保证绝无 BUG,该工具封装自 RocketMQ 团队的 Benchmark,且经过本人测试。 2 功能强大,覆盖普通消息、定时(延时)消息、事务消息(提交、回滚)等基本场景的发送场景。集群订阅和广播订阅的消费场景。内含多种...
阿里RocketMQ_用户指南_V3.2.4_最新版本.pdf benchmark.pdf Metaq在JDk 7下的异常及解决方案.docx mqvsmq.pdf RocketMQ_原理简介.pdf RocketMQ_admin.pdf RocketMQ_benchmark.pdf RocketMQ_calvinzhan - 类图.pdf ...
JAVA-ACE-架构师系列视频教程RocketMQ订单实战上目录介绍: 1001_RocketMQ_简介 2002_RocketMQ_核心概念详解 3003_RocketMQ_集群构建模型详解(一) 4004_RocketMQ_集群构建模型详解(二) 5005_RocketMQ_双主模式...
RocketMQ-Connect 二次开发源码。RocketMQ-Connect 二次开发源码。RocketMQ-Connect 二次开发源码。RocketMQ-Connect 二次开发源码。RocketMQ-Connect 二次开发源码。RocketMQ-Connect 二次开发源码。RocketMQ-...
RocketMQ源码分析,分为存储篇、NameServer篇、Broker篇、Producer篇、Consumer篇五大部分进行源码级的讲解。大致如下: 1、讲解commitlog、consumequeue、index、transaction文件等数据结构、数据读写、HA高可用等...
伴随着互联网的飞速发展, 特别是在近几年中, 移动... 本文介绍了MQTT 协议与RocketMQ 的这种开源项目的应用, 并通过RocketMQ 与Mosquitto 相结合的方式, 实现了一种基于RocketMQ 的MQTT 消息推送服务器及其分布式部署.
1.3 RocketMQ的前世今⽣ 1.4 各种MQ产品的⽐较 2 快速⼊⻔ 2.1 准备⼯作 2.2 安装RocketMQ 2.3 启动RocketMQ 2.4 测试RocketMQ 2.5 关闭RocketMQ 3 消息样例 3.1普通消息 3.1.1消息发送 3.1.2消费消息 3.2 顺序消息 ...
一、rocketmq入门到精通视频教程目录大纲 001-001_RocketMQ_简介 002-002_RocketMQ_核心概念详解 003-003_RocketMQ_集群构建模型详解(一) 004-004_RocketMQ_集群构建模型详解(二) 005-005_RocketMQ_双主模式集群...
rocketmq客户端,直接在浏览器中输入地址即可访问。这是个springboot项目,根据实际需要修改对应配置;主要是修改端口号及对应rocketMQ连接地址。
rocketmq总结
消息队列 RocketMQ 版是基于 Apache RocketMQ 构建的低延迟、高并发、高可用、高可靠的分布式 消息中间件。消息队列 RocketMQ 版既可为分布式应用系统提供异步解耦和削峰填谷的能力,同时也 具备互联网应用所需的...
RocketMq操作手册:针对入门初中级别。 消息队列 RocketMQ 是阿里巴巴集团基于高可用分布式集群技术,自主研发的云正式商用的专业消息中间件,既可为分布式应用系统提供异步解耦 和削峰填谷的能力,同时也具备互联网...
1.此jar包是根据rocketmq-console-ng工程maven编译而来 2.有两个jar包(rocketmq-console-ng-1.0.0-gbk.jar、rocketmq-console-ng-1.0.0-utf8.jar)分别对应传送的消息编码格式为gbk和utf8(方便查询消息时中文乱码...
rocketmq图形化界面,下载后运行即可
RocketMQ-Console是RocketMQ项目的扩展插件,是一个图形化管理控制台,提供Broker集群状态查看,Topic管理,Producer、Consumer状态展示,消息查询等常用功能,这个功能在安装好RocketMQ后需要额外单独安装、运行。...
rocketmq可视化控制台最新版
rocketmq设置开机启动脚本