从0到1实现完整的微服务框架-RocketMQ

 1# 所属集群名字
 2brokerClusterName=DefaultCluster
 3
 4# broker 名字，注意此处不同的配置文件填写的不一样，如果在 broker-a.properties 使用: broker-a,
 5# 在 broker-b.properties 使用: broker-b
 6brokerName=broker-a
 7
 8# 0 表示 Master，> 0 表示 Slave
 9brokerId=0
10
11# nameServer地址，分号分割
12# namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
13
14# 启动IP,如果 docker 报 com.alibaba.rocketmq.remoting.exception.RemotingConnectException: connect to <192.168.0.120:10909> failed
15# 解决方式1 加上一句 producer.setVipChannelEnabled(false);，解决方式2 brokerIP1 设置宿主机IP，不要使用docker 内部IP
16brokerIP1=192.168.0.2 #如果没有报错就不用管
17
18# 在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
19defaultTopicQueueNums=4
20
21# 是否允许 Broker 自动创建 Topic，建议线下开启，线上关闭 ！！！这里仔细看是 false，false，false
22autoCreateTopicEnable=true
23
24# 是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
25autoCreateSubscriptionGroup=true
26
27# Broker 对外服务的监听端口
28listenPort=10911
29
30# 删除文件时间点，默认凌晨4点
31deleteWhen=04
32
33# 文件保留时间，默认48小时
34fileReservedTime=120
35
36# commitLog 每个文件的大小默认1G
37mapedFileSizeCommitLog=1073741824
38
39# ConsumeQueue 每个文件默认存 30W 条，根据业务情况调整
40mapedFileSizeConsumeQueue=300000
41
42# destroyMapedFileIntervalForcibly=120000
43# redeleteHangedFileInterval=120000
44# 检测物理文件磁盘空间
45diskMaxUsedSpaceRatio=88
46# 存储路径
47# storePathRootDir=/home/ztztdata/rocketmq-all-4.1.0-incubating/store
48# commitLog 存储路径
49# storePathCommitLog=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/commitlog
50# 消费队列存储
51# storePathConsumeQueue=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/consumequeue
52# 消息索引存储路径
53# storePathIndex=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/index
54# checkpoint 文件存储路径
55# storeCheckpoint=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/checkpoint
56# abort 文件存储路径
57# abortFile=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/abort
58# 限制的消息大小
59maxMessageSize=65536
60
61# flushCommitLogLeastPages=4
62# flushConsumeQueueLeastPages=2
63# flushCommitLogThoroughInterval=10000
64# flushConsumeQueueThoroughInterval=60000
65
66# Broker 的角色
67# - ASYNC_MASTER 异步复制Master
68# - SYNC_MASTER 同步双写Master
69# - SLAVE
70brokerRole=ASYNC_MASTER
71
72# 刷盘方式
73# - ASYNC_FLUSH 异步刷盘
74# - SYNC_FLUSH 同步刷盘
75flushDiskType=ASYNC_FLUSH
76
77# 发消息线程池数量
78# sendMessageThreadPoolNums=128
79# 拉消息线程池数量
80# pullMessageThreadPoolNums=128

 1version: '3.5'
 2services:
 3  rmqnamesrv:
 4    image: foxiswho/rocketmq:server
 5    container_name: rmqnamesrv
 6    ports:
 7      - 9876:9876
 8    volumes:
 9      - ./data/logs:/opt/logs
10      - ./data/store:/opt/store
11    networks:
12        rmq:
13          aliases:
14            - rmqnamesrv
15
16  rmqbroker:
17    image: foxiswho/rocketmq:broker
18    container_name: rmqbroker
19    ports:
20      - 10909:10909
21      - 10911:10911
22    volumes:
23      - ./data/logs:/opt/logs
24      - ./data/store:/opt/store
25      - ./data/brokerconf/broker.conf:/etc/rocketmq/broker.conf
26    environment:
27        NAMESRV_ADDR: "rmqnamesrv:9876"
28        JAVA_OPTS: " -Duser.home=/opt"
29        JAVA_OPT_EXT: "-server -Xms128m -Xmx128m -Xmn128m"
30    command: mqbroker -c /etc/rocketmq/broker.conf
31    depends_on:
32      - rmqnamesrv
33    networks:
34      rmq:
35        aliases:
36          - rmqbroker
37
38  rmqconsole:
39    image: styletang/rocketmq-console-ng
40    container_name: rmqconsole
41    ports:
42      - 8080:8080
43    environment:
44        JAVA_OPTS: "-Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false"
45    depends_on:
46      - rmqnamesrv
47    networks:
48      rmq:
49        aliases:
50          - rmqconsole
51
52networks:
53  rmq:
54    name: rmq
55    driver: bridge

1docker-compose up

1public enum SendStatus {
2	SEND_ OK; // 成功
3	FLUSH_ DISK_ TIMEOUT;
4    FLUSH_ SLAVE TIMEOUT;
5	SLAVE_ NOT_ AVAILABLE;
6}

1msg.setDelayLevel(2);
2SendResult sendResult = producer.send(msg)

 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"os"
 7	"strconv"
 8
 9	"github.com/apache/rocketmq-client-go/v2"
10	"github.com/apache/rocketmq-client-go/v2/primitive"
11	"github.com/apache/rocketmq-client-go/v2/producer"
12)
13
14// Package main implements a simple producer to send message.
15func main() {
16	p, err := rocketmq.NewProducer(
17		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.2:9876"})),
18		producer.WithRetry(2),
19	)
20
21	if err != nil {
22		fmt.Printf("new producer error: %s", err.Error())
23		os.Exit(1)
24	}
25	err = p.Start()
26	if err != nil {
27		fmt.Printf("start producer error: %s", err.Error())
28		os.Exit(1)
29	}
30	topic := "hellomq"
31
32	for i := 0; i < 10; i++ {
33		msg := &primitive.Message{
34			Topic: topic,
35			Body:  []byte("Hello RocketMQ Go Client! " + strconv.Itoa(i)),
36		}
37		res, err := p.SendSync(context.Background(), msg)
38
39		if err != nil {
40			fmt.Printf("send message error: %s\n", err)
41		} else {
42			fmt.Printf("send message success: result=%s\n", res.String())
43		}
44	}
45	err = p.Shutdown()
46	if err != nil {
47		fmt.Printf("shutdown producer error: %s", err.Error())
48	}
49}

 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"os"
 7	"time"
 8
 9	"github.com/apache/rocketmq-client-go/v2"
10	"github.com/apache/rocketmq-client-go/v2/consumer"
11	"github.com/apache/rocketmq-client-go/v2/primitive"
12)
13
14func main() {
15	// 服务器有数据会推给回调
16	c, _ := rocketmq.NewPushConsumer(
17		// GroupName 多个实例负载均衡
18		consumer.WithGroupName("testGroup"),
19		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.2:9876"})),
20	)
21	err := c.Subscribe("hellomq", consumer.MessageSelector{}, func(ctx context.Context,
22		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
23		for i := range msgs {
24			fmt.Printf("subscribe callback: %v \n", msgs[i])
25		}
26
27		return consumer.ConsumeSuccess, nil
28	})
29	if err != nil {
30		fmt.Println(err.Error())
31	}
32	// Note: start after subscribe
33	err = c.Start()
34	if err != nil {
35		fmt.Println(err.Error())
36		os.Exit(-1)
37	}
38	time.Sleep(time.Hour)
39	err = c.Shutdown()
40	if err != nil {
41		fmt.Printf("shutdown Consumer error: %s", err.Error())
42	}
43}

 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"os"
 7
 8	"github.com/apache/rocketmq-client-go/v2"
 9	"github.com/apache/rocketmq-client-go/v2/primitive"
10	"github.com/apache/rocketmq-client-go/v2/producer"
11)
12
13func main() {
14	p, _ := rocketmq.NewProducer(
15		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
16		producer.WithRetry(2),
17	)
18	err := p.Start()
19	if err != nil {
20		fmt.Printf("start producer error: %s", err.Error())
21		os.Exit(1)
22	}
23	for i := 0; i < 10; i++ {
24		msg := primitive.NewMessage("hellomq", []byte("Hello RocketMQ Go Client!"))
25		msg.WithDelayTimeLevel(2) // 设置延迟的级别
26		res, err := p.SendSync(context.Background(), msg)
27
28		if err != nil {
29			fmt.Printf("send message error: %s\n", err)
30		} else {
31			fmt.Printf("send message success: result=%s\n", res.String())
32		}
33	}
34	err = p.Shutdown()
35	if err != nil {
36		fmt.Printf("shutdown producer error: %s", err.Error())
37	}
38
39}

 1package main
 2
 3import (
 4   "context"
 5   "fmt"
 6   "os"
 7   "strconv"
 8   "time"
 9
10   "github.com/apache/rocketmq-client-go/v2"
11   "github.com/apache/rocketmq-client-go/v2/primitive"
12   "github.com/apache/rocketmq-client-go/v2/producer"
13)
14
15type DemoListener struct {
16}
17
18func NewDemoListener() *DemoListener {
19   return &DemoListener{}
20}
21
22func (dl *DemoListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
23   fmt.Println("开始执行本地逻辑")
24   time.Sleep(time.Second * 3)
25   fmt.Println("执行本地逻辑成功")
26   //return primitive.CommitMessageState
27   //return primitive.RollbackMessageState
28   // 本地执行逻辑无缘无故失败 代码异常、宕机
29   return primitive.UnknowState
30}
31
32func (dl *DemoListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
33   fmt.Println("rocketMQ 的消息回查")
34   time.Sleep(time.Second * 4)
35   return primitive.CommitMessageState
36}
37
38func main() {
39   p, _ := rocketmq.NewTransactionProducer(
40      NewDemoListener(),
41      producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.2:9876"})),
42      producer.WithRetry(1),
43   )
44   err := p.Start()
45   if err != nil {
46      fmt.Printf("start producer error: %s\n", err.Error())
47      os.Exit(1)
48   }
49
50   for i := 0; i < 1; i++ {
51      res, err := p.SendMessageInTransaction(context.Background(),
52         primitive.NewMessage("transaction", []byte("Hello RocketMQ again "+strconv.Itoa(i))))
53
54      if err != nil {
55         fmt.Printf("send message error: %s\n", err)
56      } else {
57         fmt.Printf("send message success: result=%s\n", res.String())
58      }
59   }
60   // 回查
61   time.Sleep(5 * time.Minute)
62   err = p.Shutdown()
63   if err != nil {
64      fmt.Printf("shutdown producer error: %s", err.Error())
65   }
66}