代码拉取完成,页面将自动刷新
librocketmq
rocketmq-client-go is a lightweight wrapper around rocketmq-client-cpp, so you need install
librocketmq
first.
git clone https://github.com/apache/rocketmq-client-cpp
# By default, CFLAGS contains `/usr/local/include`
sudo mkdir -p /usr/local/include/rocketmq/
sudo cp rocketmq-client-cpp/include/* /usr/local/include/rocketmq
your could download directly or build manually:
LDFLAGS
directory(as default, /usr/local/lib/
works finely).
Please make sure your LDFLAGS
directory is under the LD_LIBRARY_PATH
.go get github.com/apache/rocketmq-client-go
import rocketmq "github.com/apache/rocketmq-client-go/core"
func SendMessagge(){
producer := rocketmq.NewProducer(config)
producer.Start()
defer producer.Shutdown()
fmt.Printf("Producer: %s started... \n", producer)
for i := 0; i < 100; i++ {
msg := fmt.Sprintf("%s-*d", *body, i)
result, err := producer.SendMessageSync(&rocketmq.Message{Topic: "test", Body: msg})
if err != nil {
fmt.Println("Error:", err)
}
fmt.Printf("send message: %s result: %s\n", msg, result)
}
}
type queueSelectorByOrderID struct{}
func (s queueSelectorByOrderID) Select(size int, m *rocketmq.Message, arg interface{}) int{
return arg.(int) % size
}
type worker struct {
p rocketmq.Producer
leftMsgCount int64
}
func (w *worker) run() {
selector := queueSelectorByOrderID{}
for atomic.AddInt64(&w.leftMsgCount, -1) >= 0 {
r, err := w.p.SendMessageOrderly(
&rocketmq.Message{Topic: *topic, Body: *body}, selector, 7 /*orderID*/, 3,
)
if err != nil {
println("Send Orderly Error:", err)
}
fmt.Printf("send orderly result:%+v\n", r)
}
}
func sendMessageOrderly(config *rocketmq.ProducerConfig) {
producer, err := rocketmq.NewProducer(config)
if err != nil {
fmt.Println("create Producer failed, error:", err)
return
}
producer.Start()
defer producer.Shutdown()
wg := sync.WaitGroup{}
wg.Add(*workerCount)
workers := make([]worker, *workerCount)
for i := range workers {
workers[i].p = producer
workers[i].leftMsgCount = (int64)(*amount)
}
for i := range workers {
go func(w *worker) { w.run(); wg.Done() }(&workers[i])
}
wg.Wait()
}
func ConsumeWithPush(config *rocketmq.PushConsumerConfig) {
consumer, err := rocketmq.NewPushConsumer(config)
if err != nil {
println("create Consumer failed, error:", err)
return
}
ch := make(chan interface{})
var count = (int64)(*amount)
// MUST subscribe topic before consumer started.
consumer.Subscribe("test", "*", func(msg *rocketmq.MessageExt) rocketmq.ConsumeStatus {
fmt.Printf("A message received: \"%s\" \n", msg.Body)
if atomic.AddInt64(&count, -1) <= 0 {
ch <- "quit"
}
return rocketmq.ConsumeSuccess
})
err = consumer.Start()
if err != nil {
println("consumer start failed,", err)
return
}
fmt.Printf("consumer: %s started...\n", consumer)
<-ch
err = consumer.Shutdown()
if err != nil {
println("consumer shutdown failed")
return
}
println("consumer has shutdown.")
}
func ConsumeWithPull(config *rocketmq.PullConsumerConfig, topic string) {
consumer, err := rocketmq.NewPullConsumer(config)
if err != nil {
fmt.Printf("new pull consumer error:%s\n", err)
return
}
err = consumer.Start()
if err != nil {
fmt.Printf("start consumer error:%s\n", err)
return
}
defer consumer.Shutdown()
mqs := consumer.FetchSubscriptionMessageQueues(topic)
fmt.Printf("fetch subscription mqs:%+v\n", mqs)
total, offsets, now := 0, map[int]int64{}, time.Now()
PULL:
for {
for _, mq := range mqs {
pr := consumer.Pull(mq, "*", offsets[mq.ID], 32)
total += len(pr.Messages)
fmt.Printf("pull %s, result:%+v\n", mq.String(), pr)
switch pr.Status {
case rocketmq.PullNoNewMsg:
break PULL
case rocketmq.PullFound:
fallthrough
case rocketmq.PullNoMatchedMsg:
fallthrough
case rocketmq.PullOffsetIllegal:
offsets[mq.ID] = pr.NextBeginOffset
case rocketmq.PullBrokerTimeout:
fmt.Println("broker timeout occur")
}
}
}
var timePerMessage time.Duration
if total > 0 {
timePerMessage = time.Since(now) / time.Duration(total)
}
fmt.Printf("total message:%d, per message time:%d\n", total, timePerMessage)
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。