2jetstream实操篇

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# 起一个 jetstream server
$ nats-server --js 

# 1. 使用向导的方式来创建stream
$ nats stream add
? Stream Name ORDERS
? Subjects ORDERS.*                ORDERS.*主题的所有消息都存储到这个stream里
? Storage file
? Replication 1
? Retention Policy Limits
? Discard Policy Old
? Stream Messages Limit -1
? Per Subject Messages Limit -1
? Total Stream Size -1
? Message TTL -1
? Max Message Size -1
? Duplicate tracking time window 2m0s
? Allow message Roll-ups No
? Allow message deletion Yes
? Allow purging subjects or the entire stream Yes

# 2. 创建3个消费者
$ nats con add
? Consumer name NEW
? Delivery target (empty for Pull Consumers)
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgement policy explicit
? Replay policy instant
? Filter Stream by subject (blank for all) ORDERS.received
? Maximum Allowed Deliveries -1
? Maximum Acknowledgements Pending 0
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
? Select a Stream  [Use arrows to move, type to filter]
> ORDERS         选择stream名字

$ nats con add
? Consumer name DISPATCH
? Delivery target (empty for Pull Consumers)
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgement policy explicit
? Replay policy instant
? Filter Stream by subject (blank for all) ORDERS.processed
? Maximum Allowed Deliveries -1
? Maximum Acknowledgements Pending 0
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
? Select a Stream ORDERS

$ nats con add
? Consumer name MONITOR
? Delivery target (empty for Pull Consumers) monitor.ORDERS
? Delivery Queue Group
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgement policy none
? Replay policy instant
? Filter Stream by subject (blank for all)
? Idle Heartbeat 0s
? Enable Flow Control, ie --flow-control No
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
? Select a Stream ORDERS

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// OrderProcess.go
package main

import (
    "log"
    "fmt"
    "github.com/nats-io/nats.go"
)

func main() {
    nc, _ := nats.Connect("localhost:4222")
    js, _ := nc.JetStream()
    fmt.Println("开始处理订单...")
    sub, _ := js.PullSubscribe("ORDERS.received", "NEW")
    for {
        msgs, _ := sub.Fetch(100)
        for _, msg := range msgs {
            log.Println("正在处理新订单：" + string(msg.Data))
            // TODO 在这儿处理新订单的各种业务，处理完毕，发送一条消息
            _, err := js.Publish("ORDERS.processed", []byte(string(msg.Data)+",processed"))
            if err != nil {
                log.Println(err.Error())
                continue
            }
            log.Println("新订单处理完毕：" + string(msg.Data))
            msg.Ack()
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// OrderComplete.go
package main

import (
    "log"
    "strings"
    "github.com/nats-io/nats.go"
)

func main() {
    nc, _ := nats.Connect("localhost:4222")
    js, _ := nc.JetStream()
    sub, _ := js.PullSubscribe("ORDERS.processed", "DISPATCH")
    for {
        msgs, _ := sub.Fetch(10)
        for _, msg := range msgs {
            orderInfo := strings.Split(string(msg.Data), ",")
            orderNo := orderInfo[0]
            log.Println("处理订单：" + orderNo)
            // TODO 在这儿处理各种业务，处理结束，推送ORDERS.completed
            _, err := js.Publish("ORDERS.completed", []byte(orderNo+",completed"))
            if err != nil {
                log.Println(err.Error())
                continue
            }
            log.Println("订单处理完毕：" + orderNo)
            msg.Ack()
        }
    }
}

1
2
3
4
5
6
7
8

# 终端1
$ go run OrderProcess.go
# 终端2
$ go run OrderComplete.go
# 终端3
$ nats sub monitor.ORDERS
# 终端4
$ nats pub ORDERS.received --count=10 --sleep=500ms "1000100{{Count}}"