RocketMq消息队列使用

简介: 最近在看消息队列框架 ,alibaba的RocketMQ单机支持1万以上的持久化队列,支持诸多特性,目前RocketMQ在阿里集团被广泛应用在订单,交易,充值,流计算,消息推送,日志流式处理,binglog分发等场景比kafka还是有过之无不及,其实kafka文档很丰富但RocketMQ网上的...

最近在看消息队列框架 ,alibaba的RocketMQ单机支持1万以上的持久化队列,支持诸多特性,

目前RocketMQ在阿里集团被广泛应用在订单,交易,充值,流计算,消息推送,日志流式处理,binglog分发等场景

比kafka还是有过之无不及,其实kafka文档很丰富

但RocketMQ网上的文章太少,找不到相关的操作教程

于是研究了下源码 做个单机操作的教程,如果你也对此有兴趣不妨共同研究

下载源码的地址 https://github.com/alibaba/RocketMQ/releases

  • 首选通过在java项目里面Maven依赖方式引用RocketMQ Java SDK

    <dependency>
        <groupId>com.alibaba.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>3.2.6</version>
    </dependency>

Downloads

在linux 下用wget 下载源码然后解压出来

在runserver.sh里面可以配置 jvm启动的参数 JAVA_OPT_1="-server -Xms4g -Xmx4g -Xmn2g -XX:PermSize=128m -XX:MaxPermSize=320m"  

可以 vi runserver.sh

分别给 mqnamesrv mqbroker play.sh 执行的权限

chmod +x  mqnamersrv 

chmod +x  mqbroker 

chmod +x  play.sh 

下面红线框的这段 命令输入错误了,忽略不用看

通过 nohup sh mqnamesrv& 启动 RocketMq

目前没看到结束的命令,也没找到相关的介绍,

我这里用的 ps -ef|grep rocketmq  查到进程pid

然后kill pid号

或则pkill -9 java [慎用]

用jps -v 查看下java进程的参数

 rocketmq启动后监听 9876端口,这里还是在看源码里面看到的,资料实在是太少了

在防火墙配置里面加上 9876端口,设置iptables对外开放

部署Broker 

nohup sh mqbroker -n "127.0.0.1:9876" -c ../conf/2m-2s-async/broker-a.properties & 

这里ip换成本机的就是单机实例,如果配置主从 这里可以配其他的ip

 Master和Slave的配置文件参考conf目录下的配置文件

 Master与Slave通过指定相同的brokerName参数来配对,Master的BrokerId必须是0,Slave的BrokerId必须是大于0的数

 一个Master下面可以挂载多个Slave,同一Master下的多个Slave通过指定不同的BrokerId来区分

 部署一Master一Slave,集群采用异步复制方式:

 Master: nohup sh mqbroker -n "192.168.1.23:9876" -c ../conf/2m-2s-async/broker-a.properties &  

Slave:   nohup sh mqbroker -n "192.168.1.23:9876" -c ../conf/2m-2s-async/broker-a-s.properties &  

 

 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
package  com.pgsqlmybatis.common.rocketmq; /*
***************************************************************
* 公司名称    :
* 系统名称    :信用管家专业版
* 类 名 称    :Ios渠道idfa统计,推广统计用
* 功能描述    :
* 业务描述    :
* 作 者 名    :@Author Royal
* 开发日期    :2016-05-15
* Created     :IntelliJ IDEA
***************************************************************
* 修改日期    :
* 修 改 者    :
* 修改内容    :
***************************************************************
*/
 
import  com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import  com.alibaba.rocketmq.client.producer.SendResult;
import  com.alibaba.rocketmq.common.message.Message;
 
public  class  Producer {
     public  static  void  main(String[] args) {
         DefaultMQProducer producer =  new  DefaultMQProducer( "Producer" );
         producer.setNamesrvAddr( "xxxxxxxxxx:9876" );
         try  {
             producer.start();
 
             String pushMsg= "kafka activeMq rocketMq 消息队列使用1" ;
             Message msg =  new  Message( "PushTopic" , "push" , "1" ,
                     pushMsg.getBytes( "UTF-8" ));
 
             SendResult result = producer.send(msg);
             System.out.println( "id:"  + result.getMsgId() +
                     " result:"  + result.getSendStatus());
 
             String pushMsg2= "海量级消息记录单机测试2" ;
             msg =  new  Message( "PushTopic" , "push" , "2" ,pushMsg2.getBytes( "UTF-8" ));
 
             result = producer.send(msg);
             System.out.println( "id:"  + result.getMsgId() +
                     " result:"  + result.getSendStatus());
 
             String pushMsg3= "海量级消息记录单机测试3" ;
             msg =  new  Message( "PullTopic" , "pull" , "1" ,pushMsg3.getBytes());
 
             result = producer.send(msg);
             System.out.println( "id:"  + result.getMsgId() +
                     " result:"  + result.getSendStatus());
         catch  (Exception e) {
             e.printStackTrace();
         finally  {
             producer.shutdown();
         }
     }
}

  

启动生成者

 

启动消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
package  com.pgsqlmybatis.common.rocketmq; /*
***************************************************************
* 公司名称    :
* 系统名称    :信用管家专业版
* 类 名 称    :Ios渠道idfa统计,推广统计用
* 功能描述    :
* 业务描述    :
* 作 者 名    :@Author Royal
* 开发日期    :2016-05-15
* Created     :IntelliJ IDEA
***************************************************************
* 修改日期    :
* 修 改 者    :
* 修改内容    :
***************************************************************
*/
 
import  java.io.UnsupportedEncodingException;
import  java.util.List;
 
import  com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import  com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import  com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import  com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import  com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import  com.alibaba.rocketmq.common.message.Message;
import  com.alibaba.rocketmq.common.message.MessageExt;
 
public  class  Consumer {
     public  static  void  main(String[] args){
         DefaultMQPushConsumer consumer =
                 new  DefaultMQPushConsumer( "PushConsumer" );
         consumer.setNamesrvAddr( "xxxxxxxxxxxx:9876" );
         try  {
             consumer.subscribe( "PushTopic" "push" );
             /**
              * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
              * 如果非第一次启动,那么按照上次消费的位置继续消费
              */
             consumer.setConsumeFromWhere(
                     ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
             consumer.registerMessageListener(
                     new  MessageListenerConcurrently() {
                         public  ConsumeConcurrentlyStatus consumeMessage(
                                 List<MessageExt> list,
                                 ConsumeConcurrentlyContext Context) {
                             Message msg = list.get( 0 );
                             System.out.println(msg.toString());
                             String recString=  null ;
                             try  {
                                 recString =  new  String(msg.getBody() , "UTF-8" );
                             catch  (UnsupportedEncodingException e) {
                                 e.printStackTrace();
                             }
                             System.out.println(recString);
                             return  ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                         }
                     }
             );
             consumer.start();
         catch  (Exception e) {
             e.printStackTrace();
         }
     }
}

   

 

以上为单机实例配置

如果你遇到什么问题可以私信我,如果觉得此文对你很有帮助,点下赞推荐下额^_^ 

参考:http://blog.csdn.net/a19881029/article/details/34446629

        http://sofar.blog.51cto.com/353572/1540874

        http://blog.csdn.net/loongshawn/article/details/51086876

        RocketMq最佳实践

       《RocketMQ原理简介》

       分布式开放消息系统(RocketMQ)的原理与实践      

       《RocketMQ用户指南》

相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
4月前
|
消息中间件 存储 缓存
RocketMQ 生产者那些事
这篇文章,我们从源码的角度探寻 RocketMQ Producer 的实现机制。
RocketMQ 生产者那些事
|
7月前
|
消息中间件 Kubernetes Cloud Native
常见的消息队列RocketMQ
常见的消息队列RocketMQ
64 0
|
8月前
|
消息中间件 数据可视化 API
RocketMQ极简入门-RocketMQ普通消息发送
RocketMQ已经写了两章了,一章是RocketMQ认识和安装,一章是RocketMQ的工作流程和核心概念,本章我们开始使用RocketMQ来发送和接收消息。RocketMQ的消息种类非常多,比如:普通消息,顺序消息,延迟消息,批量发送,消息过滤等等。本篇文章来探讨一下 普通消息的发送
193 1
|
8月前
|
消息中间件 数据可视化 API
三.RocketMQ极简入门-RocketMQ普通消息发送
RocketMQ极简入门-RocketMQ普通消息发送
|
8月前
|
消息中间件 NoSQL Java
消息队列、RabbitMQ和RocketMQ
消息队列、RabbitMQ和RocketMQ
|
消息中间件 存储 RocketMQ
消息队列RocketMQ
消息队列RocketMQ
155 0
|
存储 消息中间件 缓存
|
消息中间件 存储 负载均衡
消息队列系列4 - 原理初探之RocketMQ
RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件,前身是MetaQ,是阿里参考Kafka特点研发的一个队列模型的消息中间件,后开源给apache基金会成为了apache的顶级开源项目,具有高性能、高可靠、高实时、分布式特点。
292 0
消息队列系列4 - 原理初探之RocketMQ
|
消息中间件 中间件 RocketMQ
分布式消息队列RocketMQ
说到分布式事务,就会谈到那个经典的”账号转账”问题:2个账号,分布处于2个不同的DB,或者说2个不同的子系统里面,A要扣钱,B要加钱,如何保证原子性? 一般的思路都是通过消息中间件来实现“最终一致性”:A系统扣钱,然后发条消息给中间件,B系统接收此消息,进行加钱。 但这里面有个问题:A是先update DB,后发送消息呢? 还是先发送消息,后update DB? 假设先update DB成功,发送消息网络失败,重发又失败,怎么办? 假设先发送消息成功,update DB失败。消息已经发出去了,又不能撤回,怎么办? 所以,这里下个结论: 只要发送消息和update DB这2个操作不是原子
140 0
|
消息中间件 Java RocketMQ
RocketMQ消息队列的最佳实践(中)
RocketMQ消息队列的最佳实践
228 0
RocketMQ消息队列的最佳实践(中)

热门文章

最新文章

相关产品

  • 云消息队列 MQ