Linux的centos7.2部署rocketMq3.5.8

Stella981
• 阅读 385

背景

最近公司的终端机需要接通推送,然后查看对比了一下,觉得rocketMq更合适一点,所以选择了rocketMq进行测试。

软件清单

a、git(yum intsall git);b、jdk1.8; c、maven; d、rocketmq

操作步骤

1、首先下载rocketMq3.5.8

2、tar -zxvf rocketmq-3.5.8.tar.gz

3、进入rocketmq-3.5.8

4、sh install.sh,初始化rocketmq(这是初始化和下载东西,是必须项目)

5、下载完毕,执行 cd devenv(进入这个文件夹devenv)

6、编写rocket环境 (1)"ROCKETMQ_HOME=pwd" >> ~/.bash_profile

(2)source ~/.bash_profile

7、修改内存大小bin文件夹下面的runserver.sh文件夹的内存,从4G修改成128m和256m(因为我的测试服务器内存不大)

8、修改内存大小bin文件夹下面的runbroker.sh件夹的内存,从4G修改成128m和256m(因为我的测试服务器内存不大) 9、初始化并且授予权限

(1)cd /usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/bin

(2)授予执行权限 chmod +x mqadmin mqbroker mqfiltersrv mqnamesrv mqshutdown

10、运行namesrv服务(同时将日志归类)

(1)nohup mqnamesrv 1>/usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/log/ng.log 2>/usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/log/ng-err.log &

(2)查看ng.log,如果出现 The Name Server boot success. serializeType=JSON,则表示nameServer开启成功

11、启动mqbroker

(1)nohup sh mqbroker -n ip:9876 autoCreateTopicEnable=true > /usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/log/broker.log 2>&1 &

(2)查看broker.log,如果出现 The Name Server boot success. serializeType=JSON,则表示nameServer开启成功

12、查看是否运行成功 ps aux | grep java

如果代码里面生产者和消费者加上这句话,就走端口10909,不加就走vip的10911,看服务器日志启动哪个端口而定加不加 producer.setVipChannelEnabled(false);

代码区

1、消费者代码

package cn.shopin.ssm.testRocketMq;

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.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Administrator on 2017/12/26.
 */
public class ConsumerTest01 {
    /**
     * 当前例子是PushConsumer用法,使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br>
     * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息,然后再回调用户Listener方法<br>
     */
    public static void main(String[] args) {

        /**
         * 注意:ConsumerGroupName需要由应用来保证唯一
         */
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("ProducerGroupName");
        //pushConsumer.setNamesrvAddr("192.168.180.1:9876");
        pushConsumer.setNamesrvAddr("ip:9876");
        pushConsumer.setInstanceName("Consumer");
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
         * 如果非第一次启动,那么按照上次消费的位置继续消费
         */
        pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
            /**
             * 订阅指定topic下tags分别等于TagA或TagC或TagD
             * 两个参数:第一个参数是topic第二个参数是tags
             */
            pushConsumer.subscribe("TopicTest1", "TagA || TagC || TagD");
            pushConsumer.subscribe("TopicTest2", "*");
            pushConsumer.subscribe("TopicTest3", "*");
            /**
             * 订阅指定topic下所有消息<br>
             * 注意:一个consumer对象可以订阅多个topic
             */
            //pushConsumer.subscribe("TopicTest2", "*");
            final AtomicInteger count = new AtomicInteger();
            pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs.size());
                    MessageExt messageExt = msgs.get(0);
                    count.getAndIncrement();
                    if ("TopicTest1".equals(messageExt.getTopic())) {
                        // 执行TopicTest1的消费逻辑
                        if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
                            // 执行TagA的消费
                            System.out.println("TopicTest1的TagA:"+new String(messageExt.getBody()));
                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
                            System.out.println("TopicTest1的TagB:"+new String(messageExt.getBody()));
                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) {
                            System.out.println("TopicTest1的TagC:"+new String(messageExt.getBody()));
                        }
                    } else if ("TopicTest2".equals(messageExt.getTopic())) {
                        // 执行TopicTest1的消费逻辑
                        if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
                            // 执行TagA的消费
                            System.out.println("TopicTest2的TagA:"+new String(messageExt.getBody()));
                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
                            System.out.println("TopicTest2的TagB:"+new String(messageExt.getBody()));
                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) {
                            System.out.println("TopicTest2的TagC:"+new String(messageExt.getBody()));
                        }
                    } else if ("TopicTest3".equals(messageExt.getTopic())) {
                        // 执行TopicTest1的消费逻辑
                        if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
                            // 执行TagA的消费
                            System.out.println("TopicTest3的TagA:"+new String(messageExt.getBody()));
                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
                            System.out.println("TopicTest3的TagB:"+new String(messageExt.getBody()));
                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) {
                            System.out.println("TopicTest3的TagC:"+new String(messageExt.getBody()));
                        }
                    }
                    System.out.println("当前的count的值为:"+count);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        /**
         * Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br>
         */
        try {
            pushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        System.out.println("Consumer Started.");
    }
}

2、生产者代码

package cn.shopin.ssm.testRocketMq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2017/12/26.
 */
public class ProducerTest01 {
    public static void main(String[] args) {
        /**
         * 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br>
         * 注意:ProducerGroupName需要由应用来保证唯一<br>
         * ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键,
         * 因为服务器会回查这个Group下的任意一个Producer
         */
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        //producer.setNamesrvAddr("192.168.180.1:9876");
        producer.setNamesrvAddr("ip:9876");
        producer.setInstanceName("Producer");
        //producer.setVipChannelEnabled(false);
        /**
         * Producer对象在使用之前必须要调用start初始化,初始化一次即可<br>
         * 注意:切记不可以在每次发送消息时,都调用start方法
         */
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        for (int i = 0; i < 10; i++) {
            try {
                /**
                 * 下面这段代码表明一个Producer对象可以发送多个topic,多个tag的消息。
                 * 注意:send方法是同步调用,只要不抛异常就标识成功。但是发送成功也可会有多种状态,<br>
                 * 例如消息写入Master成功,但是Slave不成功,这种情况消息属于成功,但是对于个别应用如果对消息可靠性要求极高,<br>
                 * 需要对这种情况做处理。另外,消息可能会存在发送失败的情况,失败重试由应用来处理。
                 */
                {
                    Message msg = new Message("TopicTest1",// topic
                            "TagA",// tag
                            "OrderID001",// key
                            ("按一下").getBytes());// body
                    SendResult sendResult = producer.send(msg);
                    System.out.println("按一下:"+sendResult);
                }

                {
                    Message msg = new Message("TopicTest2",
                            "TagB",
                            "OrderID001",
                            ("Hello MetaQ TagB".getBytes()));

                    SendResult sendResult = producer.send(msg);
                    System.out.println("Hello MetaQ TagB"+sendResult);
                }

                {
                    Message msg = new Message("TopicTest3",
                            "TagC",
                            "OrderID001",
                            ("Hello MetaQ TagC").getBytes());

                    SendResult sendResult = producer.send(msg);
                    System.out.println("Hello MetaQ TagC"+sendResult);
                }

                TimeUnit.MILLISECONDS.sleep(1000);

            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (RemotingException e) {
                e.printStackTrace();
            } catch (MQBrokerException e) {
                e.printStackTrace();
            }
        }
        /**
         * 应用退出时,要调用shutdown来清理资源,关闭网络连接,从MetaQ服务器上注销自己
         * 注意:我们建议应用在JBOSS、Tomcat等容器的退出销毁方法里调用shutdown方法
         */
        producer.shutdown();
    }
}
点赞
收藏
评论区
推荐文章
秃头王路飞 秃头王路飞
4个月前
webpack5手撸vue2脚手架
webpack5手撸vue相信工作个12年的小伙伴们在面试的时候多多少少怕被问到关于webpack方面的知识,本菜鸟最近闲来无事,就尝试了手撸了下vue2的脚手架,第一次发帖实在是没有经验,望海涵。languageJavaScript"name":"vuecliversion2","version":"1.0.0","desc
浅梦一笑 浅梦一笑
4个月前
初学 Python 需要安装哪些软件?超级实用,小白必看!
编程这个东西是真的奇妙。对于懂得的人来说,会觉得这个工具是多么的好用、有趣,而对于小白来说,就如同大山一样。其实这个都可以理解,大家都是这样过来的。那么接下来就说一下python相关的东西吧,并说一下我对编程的理解。本人也是小白一名,如有不对的地方,还请各位大神指出01名词解释:如果在编程方面接触的比较少,那么对于软件这一块,有几个名词一定要了解,比如开发环
技术小男生 技术小男生
4个月前
linux环境jdk环境变量配置
1:编辑系统配置文件vi/etc/profile2:按字母键i进入编辑模式,在最底部添加内容:JAVAHOME/opt/jdk1.8.0152CLASSPATH.:$JAVAHOME/lib/dt.jar:$JAVAHOME/lib/tools.jarPATH$JAVAHOME/bin:$PATH3:生效配置
光头强的博客 光头强的博客
4个月前
Java面向对象试题
1、请创建一个Animal动物类,要求有方法eat()方法,方法输出一条语句“吃东西”。创建一个接口A,接口里有一个抽象方法fly()。创建一个Bird类继承Animal类并实现接口A里的方法输出一条有语句“鸟儿飞翔”,重写eat()方法输出一条语句“鸟儿吃虫”。在Test类中向上转型创建b对象,调用eat方法。然后向下转型调用eat()方
刚刚好 刚刚好
4个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
小森森 小森森
4个月前
校园表白墙微信小程序V1.0 SayLove -基于微信云开发-一键快速搭建,开箱即用
后续会继续更新,敬请期待2.0全新版本欢迎添加左边的微信一起探讨!项目地址:(https://www.aliyun.com/activity/daily/bestoffer?userCodesskuuw5n)\2.Bug修复更新日历2.情侣脸功能大家不要使用了,现在阿里云的接口已经要收费了(土豪请随意),\\和注意
晴空闲云 晴空闲云
4个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
3个月前
快速入门|使用MemFire Cloud构建React Native应用程序
MemFireCloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
helloworld_28799839 helloworld_28799839
4个月前
常用知识整理
Javascript判断对象是否为空jsObject.keys(myObject).length0经常使用的三元运算我们经常遇到处理表格列状态字段如status的时候可以用到vue