1 简介

阿里开源, Java和Scala开发, 2016年后由Apache基金会维护

2 安装

(1) docker-compose.yml

version: '3.5'
services:
  rmqnamesrv:
    image: foxiswho/rocketmq:server
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    volumes:
      - ./logs:/opt/logs
      - ./store:/opt/store
    command: sh mqnamesrv
    networks:
      rmq:
        aliases:
          - rmqnamesrv
  rmqbroker:
    image: foxiswho/rocketmq:broker
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
    volumes:
        - ./logs:/opt/logs
      - ./store:/opt/store
      - ./conf/broker.conf:/etc/rocketmq/broker.conf
    command: sh mqbroker  -c /opt/rocketmq-4.4.0/conf/broker.conf
    environment:
      NAMESRV_ADDR: "rmqnamesrv:9876"
      JAVA_OPTS: " -Duser.home=/opt"
      JAVA_OPT_EXT: "-server -Xms256m -Xmx256m -Xmn256m"
    command: mqbroker -c /etc/rocketmq/broker.conf
    depends_on:
      - rmqnamesrv
    networks:
      rmq:
        aliases:
          - rmqbroker
  rmqconsole:
    image: styletang/rocketmq-console-ng
    container_name: rmqconsole
    restart: always
    ports:
      - 8080:8080
    environment:
      JAVA_OPTS: "-Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false"
    depends_on:
      - rmqnamesrv
    volumes:
      - ./console/logs:/root/logs
    networks:
      rmq:
        aliases:
          - rmqconsole
networks:
  rmq:
    name: rmq
    driver: bridge

(2) ./conf/broker.conf

# nameServer地址，如果nameserver是多台集群的话，就用分号分割
namesrvAddr=192.168.1.103:9876
# 所属集群名字(同一主从下:Master和slave名称要一致)
brokerClusterName=DefaultCluster
# broker名字，注意此处不同的配置文件填写的不一样  例如：在a.properties 文件中写 broker-a  在b.properties 文件中写 broker-b
brokerName=broker-a
# 0 表示 Master，>0 表示 Slave
brokerId=0
# 启动IP, 要设置为其它机器能访问的IP
brokerIP1=192.168.1.103
# Broker 对外服务的监听端口
listenPort=10911
# 在发送消息时，自动创建服务器不存在的topic，默认创建的队列数。由于是4个broker节点，所以设置为4
# defaultTopicQueueNums=4
# 是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
# 是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
# commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
# ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
# 检测可用的磁盘空间大小,当磁盘被占用超过95%，消息写入会直接报错                    
diskMaxUsedSpaceRatio=95

(3) docker-compose up

3 管理后台

http://192.168.1.103:8080/

• 添加topic

4 消息类型

(1) 普通消息(订阅)

(2) 顺序消息

(3) 延时消息

(4) 事务消息

5 搭建py开发环境

https://github.com/apache/rocketmq-client-python

(1) 安装librocketmq

wget https://github.com/apache/rocketmq-client-cpp/releases/download/2.0.0/rocketmq-client-cpp-2.0.0.amd64.deb sudo dpkg -i rocketmq-client-cpp-2.0.0.amd64.deb

(2) pip install rocketmq-client-python

(3) producer.py

from rocketmq.client import Producer, Message
producer = Producer('PID-XXX')
producer.set_name_server_address('127.0.0.1:9876')
producer.start()
msg = Message('YOUR-TOPIC')
msg.set_keys('XXX')
msg.set_tags('XXX')
msg.set_body('XXXX')
# 延时消息, 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
# msg.set_delay_time_level(3)
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
producer.shutdown()

(4) consumer.py

from rocketmq.client import PushConsumer, ConsumeStatus
def callback(msg):
    print(msg.id, msg.body)
    return ConsumeStatus.CONSUME_SUCCESS
consumer = PushConsumer('CID_XXX')
consumer.set_name_server_address('127.0.0.1:9876')
consumer.subscribe('YOUR-TOPIC', callback)
consumer.start()
import time
time.sleep(2)
consumer.shutdown()

0 配置属性详解