单节点rocketmq环境搭建

忘是亡心i 2021-11-17 08:06 327阅读 0赞

官网地址: [http://rocketmq.apache.org/docs/quick-start/][http_rocketmq.apache.org_docs_quick-start]

下载源码地址: [https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.4.0/rocketmq-all-4.4.0-source-release.zip][https_www.apache.org_dyn_closer.cgi_path_rocketmq_4.4.0_rocketmq-all-4.4.0-source-release.zip]

**环境准备:**

jdk1.8的安装参考: [https://blog.csdn.net/qq\_38270106/article/details/83048876][https_blog.csdn.net_qq_38270106_article_details_83048876]

maven的安装参考: [https://blog.csdn.net/qq\_38270106/article/details/97764483][https_blog.csdn.net_qq_38270106_article_details_97764483]

### 1 下载好压缩包之后使用rz目录上传到/usr/local目录, 解压 ###

> cd /usr/local
> 
> rz
> 
> unzip rocketmq-all-4.4.0-source-release.zip

### 2 打包 ###

> cd /usr/local/rocketmq-all-4.4.0/
> 
> mvn -Prelease-all -DskipTests clean install -U

### 3修改配置(如果不修改可能启动报错内存溢出) ###

> cd /usr/local/rocketmq-all-4.4.0/distribution/target/apache-rocketmq

3.1修改runserver.sh

> vi bin/runserver.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn125m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

3.2修改runbroker.sh

> vi bin/runbroker.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn125m"

3.3 修改broker.conf

> namesrvAddr = 127.0.0.1:9876  
> brokerIP1 = 192.168.44.135 \#这里改成虚拟机ip

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70][]

### 4 启动namesrv和broker ###

> nohup sh bin/mqnamesrv &
> 
> nohup sh bin/mqbroker -n localhost:9876 autoCreateTopicEnable=true -c conf/broker.conf &

查看启动log

> cat nohup.log

![20190730130350491.png][]

###  5 代码测试 ###

pom.xml

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.0.0-incubating</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.58</version>
            </dependency>

生产者

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.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    /**
     * @description:
     * @author: liuwq
     * @date: 2019/7/29 0029 下午 3:30
     * @version: V1.0
     */
    public class Producer {
    
        public static void main(String[] args) throws MQClientException, InterruptedException {
    
            //声明并初始化一个producer
            //需要一个producer group名字作为构造方法的参数，这里为producer1
            DefaultMQProducer producer = new DefaultMQProducer("producer1");
    
            //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
            //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
            producer.setNamesrvAddr("192.168.44.135:9876");
    
            //调用start()方法启动一个producer实例
            producer.start();
    
            //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
            for (int i = 0; i < 10; i++) {
                try {
                    Message msg = new Message("TopicTest",// topic
                            "TagA",// tag
                            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                    );
    
                    //调用producer的send()方法发送消息
                    //这里调用的是同步的方式，所以会有返回结果
                    SendResult sendResult = producer.send(msg);
    
                    //打印返回结果，可以看到消息发送的状态以及一些相关信息
                    System.out.println(sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
    
            //发送完消息之后，调用shutdown()方法关闭producer
            producer.shutdown();
        }
    
    }

消费者

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.io.UnsupportedEncodingException;
    import java.util.List;
    
    /**
     * @description:
     * @author: liuwq
     * @date: 2019/7/29 0029 下午 3:31
     * @version: V1.0
     */
    public class Consumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
    
            //声明并初始化一个consumer
            //需要一个consumer group名字作为构造方法的参数，这里为consumer1
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");
    
            //同样也要设置NameServer地址 多个地址用分号隔开
            consumer.setNamesrvAddr("192.168.44.135:9876");
    
            //这里设置的是一个consumer的消费策略
            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
            consumer.subscribe("TopicTest", "*");
    
            //设置一个Listener，主要进行消息的逻辑处理
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
    
                    System.out.println(Thread.currentThread().getName() + " 收到新消息: ");
                    for(MessageExt me:msgs){
                        try {
                            System.out.println(new String(me.getBody(),"utf-8"));
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
    
                    //返回消费状态
                    //CONSUME_SUCCESS 消费成功
                    //RECONSUME_LATER 消费失败，需要稍后重新消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            //调用start()方法启动consumer
            consumer.start();
    
            System.out.println("Consumer Started.");
        }
    
    }

###  6 停止namesrv和borker(可不执行) ###

> sh bin/mqshutdown broker
> 
> sh bin/mqshutdown namesrv

###  7 rocketmq-console安装 ###

github下载源码地址: [https://github.com/apache/rocketmq-externals][https_github.com_apache_rocketmq-externals]

下载完之后使用rz目录上传到/usr/local目录, 解压

> cd /usr/local
> 
> rz
> 
> unzip rocketmq-externals-master.zip

### 8 修改配置文件namesrv地址 ###

> vi /usr/local/rocketmq-externals-master/rocketmq-console/src/main/resources/application.properties

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 1][]

### 9 maven打包 ###

> cd /usr/local/rocketmq-externals-master/rocketmq-console
> 
> mvn clean install -DskipTests

### 10 运行jar包  ###

> java -jar /usr/local/rocketmq-externals-master/rocketmq-console/target/rocketmq-console-ng-1.0.1.jar

访问: [http://192.168.44.135:8080/\#/][http_192.168.44.135_8080]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 2][]

[http_rocketmq.apache.org_docs_quick-start]: http://rocketmq.apache.org/docs/quick-start/
[https_www.apache.org_dyn_closer.cgi_path_rocketmq_4.4.0_rocketmq-all-4.4.0-source-release.zip]: https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.4.0/rocketmq-all-4.4.0-source-release.zip
[https_blog.csdn.net_qq_38270106_article_details_83048876]: https://blog.csdn.net/qq_38270106/article/details/83048876
[https_blog.csdn.net_qq_38270106_article_details_97764483]: https://blog.csdn.net/qq_38270106/article/details/97764483
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70]: /images/20211116/52e33d87fb6f43ffbc830e5e193d7796.png
[20190730130350491.png]: /images/20211116/3eb70a42f228471ba878fa13c915a4c0.png
[https_github.com_apache_rocketmq-externals]: https://github.com/apache/rocketmq-externals
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 1]: /images/20211116/bb259b02dcd84f4785cc6d5f4093fb2b.png
[http_192.168.44.135_8080]: http://192.168.44.135:8080/#/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4MjcwMTA2_size_16_color_FFFFFF_t_70 2]: /images/20211116/c23e842beb93451585d8ce4ed90f4914.png