最簡消息隊列的實現(xiàn)方法

這篇文章主要介紹“最簡消息隊列的實現(xiàn)方法”，在日常操作中，相信很多人在最簡消息隊列的實現(xiàn)方法問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”最簡消息隊列的實現(xiàn)方法”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

創(chuàng)新互聯(lián)是一家集網(wǎng)站制作、成都網(wǎng)站設計、網(wǎng)站頁面設計、網(wǎng)站優(yōu)化SEO優(yōu)化為一體的專業(yè)網(wǎng)絡公司，已為成都等多地近百家企業(yè)提供網(wǎng)站建設服務。追求良好的瀏覽體驗,以探求精品塑造與理念升華,設計最適合用戶的網(wǎng)站頁面。 合作只是第一步，服務才是根本,我們始終堅持講誠信，負責任的原則，為您進行細心、貼心、認真的服務,與眾多客戶在蓬勃發(fā)展的市場環(huán)境中,互促共生。

使用

結合其他 mq 的使用經(jīng)歷，基本的使用流程：

1. 創(chuàng)建 producer 或 consumer

2. 啟動 mq

3. 生產消息/消費消息

對應到 queue 中，大致也是這個：

創(chuàng)建 queue

// 生產者創(chuàng)建工廠
producer := newMockedProducer()
// 消費者創(chuàng)建工廠
consumer := newMockedConsumer()
// 將生產者以及消費者的創(chuàng)建工廠函數(shù)傳遞給 NewQueue()
q := queue.NewQueue(func() (Producer, error) {
  return producer, nil
}, func() (Consumer, error) {
  return consumer, nil
})

我們看看 NewQueue 需要什么構建條件：

1. producer constructor

2. consumer constructor

將雙方的工廠函數(shù)傳遞給 queue ，由它去執(zhí)行以及重試。

這兩個需要的目的是將生產者/消費者的構建和消息生產/消費都封裝在 mq 中，而且將生產者/消費者的整套邏輯交給開發(fā)者處理：

type (
	// 開發(fā)者需要實現(xiàn)此接口
	Producer interface {
		AddListener(listener ProduceListener)
		Produce() (string, bool)
	}
	...
	// ProducerFactory定義了生成Producer的方法
	ProducerFactory func() (Producer, error)
)

1. 其實也就是將生產者的邏輯交個開發(fā)者自己完成，mq 只負責生產者/消費者的消息傳遞和之間的調度。

2. 工廠方法的設計，是將生產者本身和生產消息，這兩個任務都交給 queue 自己來做調度或者重試。

生產msg

生產消息當然要回到生產者本身：

type mockedProducer struct {
	total int32
	count int32
  // 使用waitgroup來模擬任務的完成
	wait  sync.WaitGroup
}
// 實現(xiàn) Producer interface 的方法：Produce()
func (p *mockedProducer) Produce() (string, bool) {
	if atomic.AddInt32(&p.count, 1) <= p.total {
		p.wait.Done()
		return "item", true
	}
	time.Sleep(time.Second)
	return "", false
}

queue 中的生產者編寫都必須實現(xiàn)：

• Produce()：由開發(fā)者編寫生產消息的邏輯

• AddListener()：生產者

消費msg

和生產者類似：

type mockedConsumer struct {
	count  int32
}

func (c *mockedConsumer) Consume(string) error {
	atomic.AddInt32(&c.count, 1)
	return nil
}

啟動 queue

啟動，然后驗證我們上述的生產者和消費者之間的數(shù)據(jù)是否傳輸成功：

func TestQueue(t *testing.T) {
	producer := newMockedProducer(rounds)
	consumer := newMockedConsumer()
	// 創(chuàng)建 queue
	q := NewQueue(func() (Producer, error) {
		return producer, nil
	}, func() (Consumer, error) {
		return consumer, nil
	})
	// 當生產者生產完畢，執(zhí)行 Stop() 關閉生產端生產
	go func() {
		producer.wait.Wait()
    // mq生產端停止生產，不是mq本身 Stop 運行
		q.Stop()
	}()
	// 啟動
	q.Start()
	// 驗證生產消費端是否消息消費完成
	assert.Equal(t, int32(rounds), atomic.LoadInt32(&consumer.count))
}

以上就是 queue 最簡易的入門使用代碼。開發(fā)者可以根據(jù)自己的業(yè)務實際情況：自由定義生產者/消費者已經(jīng)生產/消費邏輯。

整體設計


整體流程如上圖：

1. 全體的通信都由 channel 進行

2. 通過加入監(jiān)聽器 listener ，以及事件觸發(fā) event ，相當于將觸發(fā)器邏輯分離出來

3. 生產者有 produceone ，這個是生產消息的邏輯，但是其中的 Produce() 是由開發(fā)者編寫【上面的 interface 中正是這個函數(shù)】

4. 同理消費者，Consume()

基本的消息流動就入上圖以及上述描寫的，具體的代碼分析我們就留到下一篇，我們????分析里面，尤其是如何控制 channel 是整個設計的核心。

到此，關于“最簡消息隊列的實現(xiàn)方法”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關知識，請繼續(xù)關注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>

標題名稱：最簡消息隊列的實現(xiàn)方法
標題來源：http://bm7419.com/article18/psdhdp.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供定制開發(fā)、軟件開發(fā)、網(wǎng)站營銷、小程序開發(fā)、靜態(tài)網(wǎng)站、ChatGPT

廣告

聲明：本網(wǎng)站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經(jīng)允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)