RocketMQ

RocketMQ ǰͰͿԴķֲʽϢм֧Ϣ˳ϢϢʱϢϢݵȡмڱ׼ϢмĸGroupTopicQueueȡϵͳProducerConsumerBrokerNameServerȡ

RocketMQ ص

ַ֧/ģPub/Sub͵Ե㣨P2PϢģͣ
һпɿȽȳFIFOϸ˳򴫵ݣRocketMQ Ա֤ϸϢ˳򣬶ActiveMQ ޷֤
֧PullƣPushϢģʽPush ⣬߶ Listener ص PullȨӦãӦҪĵϢ Broker ȡϢһλü¼⣨¼ᵼϢظѣ
һаϢĶѻRocketMQ ṩڼϢĶѻⲻص㣬صǶѻڼϢȻдӳ٣
ֶ֧ϢЭ飬 JMSMQTT ȣ
ֲʽ߿õĲܹһϢ壻RocketMQ ԭֲַ֧ʽģActiveMQ ԭڵԣ
ṩ docker ڸԺƼȺ
ṩáָͼصȹܷḻ Dashboard

Broker

Broker ʵ RocketMQ 洢ϢתϢBroker RocketMQ ϵͳиմ߷Ϣ洢ͬʱΪߵȡ׼Broker Ҳ洢ϢصԪݣ顢ѽƫƺͶϢȡ

Broker Server RocketMQ ҵģ˶Ҫģ飺

·ģ飺 Broker ʵ壬 clients ˵
ͻ˹ͻ(Producer/Consumer)ά Consumer Topic Ϣ
洢ṩ򵥵 API ӿڴϢ洢Ӳ̺Ͳѯܡ
߿÷񣺸߿÷ṩ Master Broker Slave Broker ֮ͬܡ
Ϣ񣺸ض Message key Ͷݵ Broker ϢṩϢĿٲѯ

NameServer

NameServer һǳ򵥵 Topic ·עģɫ Dubbo е zookeeper֧ Broker Ķ̬ע뷢֡

Ҫܣ

Broker NameServer Broker ȺעϢұΪ·ϢĻݡȻṩƣ Broker Ƿ񻹴
·Ϣ Producer Consumer ṩȡ Broker бÿ NameServer Broker Ⱥ·Ϣڿͻ˲ѯĶϢȻ Producer Conumser ͨ NameServer Ϳ֪ Broker Ⱥ·ϢӶϢͶݺѡ

ʹ Docker ٴ RocketMQ 4.4

rocketmq Ҫ broker nameserver ǵֿȽ鷳ｫʹ docker-compose⣬Ҫһ web ӻ̨Լ mq ״̬ԼϢʹ rocketmq-consoleͬóҲʹ docker װ

linux ѡ񲢽Ŀ¼

mkdir rocketmq-docker ƴ

rocketmq-docker Ŀ¼һΪ broker.conf ļ£

Ⱥƣڵ϶ö

brokerClusterName = DefaultCluster

brokerƣmasterslaveʹͬƣǵӹϵ

brokerName = broker-a

0ʾMaster0ʾͬslave

brokerId = 0

ʾϢɾĬ賿4

deleteWhen = 04

ڴϱϢʱλСʱ

fileReservedTime = 48

ֵSYNC_MASTERASYNC_MASTERSLAVEͬ첽ʾMasterSlave֮ͬݵĻƣ

brokerRole = ASYNC_MASTER

ˢ̲ԣȡֵΪASYNC_FLUSHSYNC_FLUSHʾͬˢ̺첽ˢ̣SYNC_FLUSHϢд̺ŷسɹ״̬ASYNC_FLUSHҪ

flushDiskType = ASYNC_FLUSH

brokerڵڷipַ

brokerIP1 = 192.168.138.131

ע⣺ brokerIP1 ãĬϻΪ docker ڲIPӲϡ

rocketmq-docker Ŀ¼һΪ rocketmq.yaml Ľűļ

rocketmq.yaml :

version: '2'
services:
  namesrv:
    image: rocketmqinc/rocketmq
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    volumes:
      - /docker/rocketmq/data/namesrv/logs:/home/rocketmq/logs
      - /docker/rocketmq/data/namesrv/store:/home/rocketmq/store
    command: sh mqnamesrv
  broker:
    image: rocketmqinc/rocketmq
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
      - 10912:10912
    volumes:
      - /docker/rocketmq/data/broker/logs:/home/rocketmq/logs
      - /docker/rocketmq/data/broker/store:/home/rocketmq/store
      - /docker/rocketmq/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf
    command: sh mqbroker -n namesrv:9876 -c /opt/rocketmq-4.4.0/conf/broker.conf
    depends_on:
      - namesrv
    environment:
      - JAVA_HOME=/usr/lib/jvm/jre
  console:
    image: styletang/rocketmq-console-ng
    container_name: rocketmq-console-ng
    ports:
      - 8087:8080
    depends_on:
      - namesrv
    environment:
      - JAVA_OPTS= -Dlogging.level.root=info   -Drocketmq.namesrv.addr=rmqnamesrv:9876 
      - Dcom.rocketmq.sendMessageWithVIPChannel=false
ƴ

broker õķǽ˿ڣʹã

firewall-cmd --zone=public --add-port=10909-10912/tcp --permanent ƴ

ִ sentinel-dashboard.yaml ű

docker-compose -f rocketmq.yaml up ƴ
rocketmq ̨ǻᷢƵͼȻʼ϶ǿյģ

http://(װRocketMQIP):8087 ƴ

ѡ Ⱥ һܿõ broker IPˣ

Ѿ RocketMQ ˵ĲھͿȥ Spring Ŀʹÿͻˡ

Spring Ŀ RocketMQ ͻ

pom ļ

    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        rocketmq-spring-boot-starter
        <version>2.0.4</version>
    </dependency>

application.yml ã

server: port: 10801

spring: application: name: (Ŀ)-service

rocketmq: name-server: (װRocketMQIP):9876 producer: group: (Ŀ)-group ƴ

½һϢ MessageProducer ΪϢ ߣ

@Service public class MessageProducer implements CommandLineRunner {

 @Resource
 private RocketMQTemplate rocketMQTemplate;

 @Override
 public void run(String... args) throws Exception {
     rocketMQTemplate.send("test-topic-1", MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
 }

} ƴ

½һϢ MessageListener ΪϢ ߣ

@Slf4j @Service @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1") public class MessageListener implements RocketMQListener<String> {
```
 @Override
 public void onMessage(String message) {
     log.info("received message: {}", message);
 }
```
} ƴ

½һ MessageProducer ĵÿࣺ

@RestController @RequestMapping public class HelloController {

 @Resource
 private MessageProducer messageProducer;

 @RequestMapping("/message")
 public void message() throws Exception {
     messageProducer.run("");
 }

Spring Ŀһ򵥵Ϣ:

GET http://localhost:10801/message Accept: / Cache-Control: no-cache ƴ

Ҳ RocketMQ Ĺ̨¿Ϣ

ʵսﲢûнһЩܴһе㷸Ժ

Topic⣩൱һ͵Ϣ Topic1 רǷҵ Topic2 רŻȯҵ񣻶 Group飩൱ڶߺߵķ飬ǵ΢Ҳߣ Group1 Ʒ΢Group2 Ƕ΢񣬵ȻҪ߷黹߷顣

GroupΪProducerGroup ConsumerGroup, ĳһߺߣһ˵ͬһΪ Groupͬһ Group һ˵ͺѵϢһġ
TopicϢ⣬һϢͣ䷢Ϣ߶ȡϢ
Queue: ΪдֶУһ˵дһ£һ¾ͻֺܶ⡣

Topic зΪ˶ QueueʵǷ/ȡϢͨСλǷϢҪָĳдĳ QueueȡϢʱҲҪָȡĳ Queueǵ˳ϢԻǵ Queue άȱֶȫôҪ Queue СΪ1еݶ Queue

ߣײ˵ ӣhttps://juejin.cn/post/6930869079217717256 Դϡ ȨСҵתϵ߻Ȩҵתע

ο

https://lijunyi.xyz/docs/SpringCloud/SpringCloud.html#_2-2-x-%E5%88%86%E6%94%AF https://mp.weixin.qq.com/s/2jeovmj77O9Ux96v3A0NtA https://juejin.cn/post/6931922457741770760 https://github.com/D2C-Cai/herring http://c.biancheng.net/springcloud

SpringCloudAlibabaRocketMQ