RocketMQ部署安装(非Docker安装)

叁歲伎倆 2023-10-07 09:18 30阅读 0赞

## 1、下载 ##

下载地址：https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip  
版本使用目前最新版：4.3.2

##    2、非Docker安装 ##

cd /haoke
    unzip rocketmq-all-4.3.2-bin-release.zip
    cd rocketmq-all-4.3.2-bin-release
    #启动nameserver
    bin/mqnamesrv
    # The Name Server boot success. serializeType=JSON 看到这个表示已经提供成功
    #启动broker
    bin/mqbroker -n 172.16.185.55:9876  #-n 指定nameserver地址和端口
    #启动出错
    Java HotSpot(TM) 64-Bit Server VM warning: INFO:
    os::commit_memory(0x00000005c0000000, 8589934592, 0) failed; error='Cannot allocate
    memory' (errno=12)
    ……………………………………………………………………………………

启动错误，是因为内存不够，导致启动失败，原因：RocketMQ的配置默认是生产环境的配置，设置的jvm的内存  
大小值比较大，对于学习而言没有必要设置这么大，测试环境的内存往往都不是很大，所以需要调整默认值。

#调整默认的内存大小参数
    cd bin/
    vim runserver.sh
    JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn128m -XX:MetaspaceSize=128m -
    XX:MaxMetaspaceSize=128m"
    cd bin/
    vim runbroker.sh
    JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn128m"
    #从新启动测试
    bin/mqbroker -n 172.16.55.185:9876
    The broker[itcast, 172.17.0.1:10911] boot success. serializeType=JSON and name
    server is 172.16.185.55:9876

下面进行发送消息测试：

export NAMESRV_ADDR=127.0.0.1:9876
    cd bin
    sh tools.sh org.apache.rocketmq.example.quickstart.Producer
    #测试结果
    SendResult [sendStatus=SEND_OK, msgId=AC110001473C7D4991AD336AEA5703E0,
    offsetMsgId=AC11000100002A9F00000000000E8580, messageQueue=MessageQueue
    [topic=TopicTest, brokerName=itcast, queueId=3], queueOffset=1323]
    SendResult [sendStatus=SEND_OK, msgId=AC110001473C7D4991AD336AEA5903E1,
    offsetMsgId=AC11000100002A9F00000000000E8634, messageQueue=MessageQueue
    [topic=TopicTest, brokerName=itcast, queueId=0], queueOffset=1323]
    SendResult [sendStatus=SEND_OK, msgId=AC110001473C7D4991AD336AEA5F03E2,
    offsetMsgId=AC11000100002A9F00000000000E86E8, messageQueue=MessageQueue
    [topic=TopicTest, brokerName=itcast, queueId=1], queueOffset=1323]
    SendResult [sendStatus=SEND_OK, msgId=AC110001473C7D4991AD336AEA6103E3,
    offsetMsgId=AC11000100002A9F00000000000E879C, messageQueue=MessageQueue
    [topic=TopicTest, brokerName=itcast, queueId=2], queueOffset=1323]
    #可以正常发送消息

测试接收消息：

sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
    #测试结果
    ConsumeMessageThread_7 Receive New Messages: [MessageExt [queueId=2, storeSize=180,
    queueOffset=1322, sysFlag=0, bornTimestamp=1544456244818,
    bornHost=/172.16.55.185:33702, storeTimestamp=1544456244819,
    storeHost=/172.17.0.1:10911, msgId=AC11000100002A9F00000000000E84CC,
    commitLogOffset=951500, bodyCRC=684865321, reconsumeTimes=0,
    preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0,
    properties={MIN_OFFSET=0, MAX_OFFSET=1325, CONSUME_START_TIME=1544456445397,
    UNIQ_KEY=AC110001473C7D4991AD336AEA5203DF, WAIT=true, TAGS=TagA}, body=[72, 101, 108,
    108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 57, 49],
    transactionId='null'}]]
    ConsumeMessageThread_6 Receive New Messages: [MessageExt [queueId=2, storeSize=180,
    queueOffset=1323, sysFlag=0, bornTimestamp=1544456244833,
    bornHost=/172.16.55.185:33702, storeTimestamp=1544456244835,
    storeHost=/172.17.0.1:10911, msgId=AC11000100002A9F00000000000E879C,
    commitLogOffset=952220, bodyCRC=801108784, reconsumeTimes=0,
    preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0,
    properties={MIN_OFFSET=0, MAX_OFFSET=1325, CONSUME_START_TIME=1544456445397,
    UNIQ_KEY=AC110001473C7D4991AD336AEA6103E3, WAIT=true, TAGS=TagA}, body=[72, 101, 108,
    108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 57, 53],
    transactionId='null'}]]
    #从结果中，可以看出，接收消息正常

## 3、编写Java代码进行测试 ##

第一步，创建zxl-rocketmq工程  
第二步，导入依赖

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
    http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
      <groupId>cn.zxl.rocketmq</groupId>
      <artifactId>zxl-rocketmq</artifactId>
      <version>1.0-SNAPSHOT</version>
      <dependencies>
        <dependency>
          <groupId>org.apache.rocketmq</groupId>
          <artifactId>rocketmq-client</artifactId>
          <version>4.3.2</version>
        </dependency>
      </dependencies>
      <build>
        <plugins>
          
          <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.2</version>
            <configuration>
              <source>1.8</source>
              <target>1.8</target>
              <encoding>UTF-8</encoding>
            </configuration>
          </plugin>
        </plugins>
      </build>
    </project>

第三步，编写代码

package cn.zxl.rocketmq;
    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;
    public class SyncProducer {
      public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
          DefaultMQProducer("test-group");
        // Specify name server addresses.
        producer.setNamesrvAddr("172.16.55.185:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
          //Create a message instance, specifying topic, tag and message body.
          Message msg = new Message("TopicTest11" /* Topic */,
            "TagA" /* Tag */,
           ("Hello RocketMQ " +i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
         );
          //Call send message to deliver message to one of brokers.
          SendResult sendResult = producer.send(msg);
          System.out.printf("%s%n", sendResult);
       }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
     }
    }

测试：

Exception in thread "main"
    org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException:
    sendDefaultImpl call timeout
    at
    org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(Defaul
    tMQProducerImpl.java:612)
    at
    org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducer
    Impl.java:1253)
    at
    org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducer
    Impl.java:1203)
    at
    org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:214
    )
    at cn.zxl.rocketmq.SyncProducer.main(SyncProducer.java:26)

测试结果会发现，发送消息会报错。  
原因是什么呢？  
仔细观察broker启动的信息：

The broker[itcast, 172.17.0.1:10911] boot success. serializeType=JSON and name server
    is 172.16.185.55:9876

会发现，broker的ip地址是172.17.0.1，那么在开发机上是不可能访问到的。  
所以，需要指定broker的ip地址。

#创建broker配置文件
    vim /haoke/rmq/rmqbroker/conf/broker.conf
    brokerIP1=172.16.55.185
    namesrvAddr=172.16.55.185:9876
    brokerName=broker_haoke_im
    #启动broker，通过 -c 指定配置文件
    bin/mqbroker -c /haoke/rmq/rmqbroker/conf/broker.conf
    The broker[itcast, 172.16.55.185:10911] boot success. serializeType=JSON and name
    server is 172.16.55.185:9876  #这样就可以进行访问了

测试：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGFuZ3h1ZWxpYW5nLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70][]

看到，可以发送消息了。问题得到解决。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGFuZ3h1ZWxpYW5nLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200211205012565.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly96aGFuZ3h1ZWxpYW5nLmJsb2cuY3Nkbi5uZXQ=,size_16,color_FFFFFF,t_70