# 拉取鏡像
docker pull rabbitmq:3.7-management

# 創(chuàng)建并運(yùn)行容器
docker run -d --name myrabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.7-management
#5672是項(xiàng)目中連接rabbitmq的端口（我這里映射的是5672），15672是rabbitmq的web管理界面端口（我映射為15672）

# 輸入網(wǎng)址http://ip:15672即可進(jìn)入rabbitmq的web管理頁(yè)面,賬戶(hù)密碼：guest / guest

package RabbitMQ

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
)

//amqp:// 賬號(hào) 密碼@地址:端口號(hào)/vhost
const MQURL = "amqp://ziyi:ziyi@10.253.50.145:5672/ziyi"

type RabbitMQ struct {
	//連接
	conn *amqp.Connection
	//管道
	channel *amqp.Channel
	//隊(duì)列名稱(chēng)
	QueueName string
	//交換機(jī)
	Exchange string
	//key Simple模式 幾乎用不到
	Key string
	//連接信息
	Mqurl string
}

//創(chuàng)建RabbitMQ結(jié)構(gòu)體實(shí)例
func NewRabbitMQ(queuename string, exchange string, key string) *RabbitMQ {
	rabbitmq := &RabbitMQ{QueueName: queuename, Exchange: exchange, Key: key, Mqurl: MQURL}
	var err error
	//創(chuàng)建rabbitmq連接
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "創(chuàng)建連接錯(cuò)誤！")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "獲取channel失敗")
	return rabbitmq
}

//斷開(kāi)channel和connection
func (r *RabbitMQ) Destory() {
	r.channel.Close()
	r.conn.Close()
}

//錯(cuò)誤處理函數(shù)
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		log.Fatalf("%s:%s", message, err)
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

//簡(jiǎn)單模式step：1。創(chuàng)建簡(jiǎn)單模式下RabbitMQ實(shí)例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
	return NewRabbitMQ(queueName, "", "")
}

//訂閱模式創(chuàng)建rabbitmq實(shí)例
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
	//創(chuàng)建rabbitmq實(shí)例
	rabbitmq := NewRabbitMQ("", exchangeName, "")
	var err error
	//獲取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "failed to connecct rabbitmq!")
	//獲取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel!")
	return rabbitmq
}

//訂閱模式生成
func (r *RabbitMQ) PublishPub(message string) {
	//嘗試創(chuàng)建交換機(jī)，不存在創(chuàng)建
	err := r.channel.ExchangeDeclare(
		//交換機(jī)名稱(chēng)
		r.Exchange,
		//交換機(jī)類(lèi)型 廣播類(lèi)型
		"fanout",
		//是否持久化
		true,
		//是否字段刪除
		false,
		//true表示這個(gè)exchange不可以被client用來(lái)推送消息，僅用來(lái)進(jìn)行exchange和exchange之間的綁定
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		nil,
	)
	r.failOnErr(err, "failed to declare an excha"+"nge")

	//2 發(fā)送消息
	err = r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			//類(lèi)型
			ContentType: "text/plain",
			//消息
			Body: []byte(message),
		})
}

//訂閱模式消費(fèi)端代碼
func (r *RabbitMQ) RecieveSub() {
	//嘗試創(chuàng)建交換機(jī)，不存在創(chuàng)建
	err := r.channel.ExchangeDeclare(
		//交換機(jī)名稱(chēng)
		r.Exchange,
		//交換機(jī)類(lèi)型 廣播類(lèi)型
		"fanout",
		//是否持久化
		true,
		//是否字段刪除
		false,
		//true表示這個(gè)exchange不可以被client用來(lái)推送消息，僅用來(lái)進(jìn)行exchange和exchange之間的綁定
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		nil,
	)
	r.failOnErr(err, "failed to declare an excha"+"nge")
	//2試探性創(chuàng)建隊(duì)列，創(chuàng)建隊(duì)列
	q, err := r.channel.QueueDeclare(
		"", //隨機(jī)生產(chǎn)隊(duì)列名稱(chēng)
		false,
		false,
		true,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	//綁定隊(duì)列到exchange中
	err = r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，這里的key要為空
		"",
		r.Exchange,
		false,
		nil,
	)
	//消費(fèi)消息
	message, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)
	go func() {
		for d := range message {
			log.Printf("Received a message:%s,", d.Body)
		}
	}()
	fmt.Println("退出請(qǐng)按 Ctrl+C")
	<-forever
}

//話(huà)題模式 創(chuàng)建RabbitMQ實(shí)例
func NewRabbitMQTopic(exchagne string, routingKey string) *RabbitMQ {
	//創(chuàng)建rabbitmq實(shí)例
	rabbitmq := NewRabbitMQ("", exchagne, routingKey)
	var err error
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "failed     to connect rabbingmq!")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

//話(huà)題模式發(fā)送信息
func (r *RabbitMQ) PublishTopic(message string) {
	//嘗試創(chuàng)建交換機(jī)，不存在創(chuàng)建
	err := r.channel.ExchangeDeclare(
		//交換機(jī)名稱(chēng)
		r.Exchange,
		//交換機(jī)類(lèi)型 話(huà)題模式
		"topic",
		//是否持久化
		true,
		//是否字段刪除
		false,
		//true表示這個(gè)exchange不可以被client用來(lái)推送消息，僅用來(lái)進(jìn)行exchange和exchange之間的綁定
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		nil,
	)
	r.failOnErr(err, "topic failed to declare an excha"+"nge")
	//2發(fā)送信息
	err = r.channel.Publish(
		r.Exchange,
		//要設(shè)置
		r.Key,
		false,
		false,
		amqp.Publishing{
			//類(lèi)型
			ContentType: "text/plain",
			//消息
			Body: []byte(message),
		})
}

//話(huà)題模式接收信息
//要注意key
//其中* 用于匹配一個(gè)單詞，#用于匹配多個(gè)單詞（可以是零個(gè)）
//匹配 表示匹配imooc.* 表示匹配imooc.hello,但是imooc.hello.one需要用imooc.#才能匹配到
func (r *RabbitMQ) RecieveTopic() {
	//嘗試創(chuàng)建交換機(jī)，不存在創(chuàng)建
	err := r.channel.ExchangeDeclare(
		//交換機(jī)名稱(chēng)
		r.Exchange,
		//交換機(jī)類(lèi)型 話(huà)題模式
		"topic",
		//是否持久化
		true,
		//是否字段刪除
		false,
		//true表示這個(gè)exchange不可以被client用來(lái)推送消息，僅用來(lái)進(jìn)行exchange和exchange之間的綁定
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		nil,
	)
	r.failOnErr(err, "failed to declare an exchange")
	//2試探性創(chuàng)建隊(duì)列，創(chuàng)建隊(duì)列
	q, err := r.channel.QueueDeclare(
		"", //隨機(jī)生產(chǎn)隊(duì)列名稱(chēng)
		false,
		false,
		true,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	//綁定隊(duì)列到exchange中
	err = r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，這里的key要為空
		r.Key,
		r.Exchange,
		false,
		nil,
	)
	//消費(fèi)消息
	message, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)
	go func() {
		for d := range message {
			log.Printf("Received a message:%s,", d.Body)
		}
	}()
	fmt.Println("退出請(qǐng)按 Ctrl+C")
	<-forever
}

//路由模式 創(chuàng)建RabbitMQ實(shí)例
func NewRabbitMQRouting(exchagne string, routingKey string) *RabbitMQ {
	//創(chuàng)建rabbitmq實(shí)例
	rabbitmq := NewRabbitMQ("", exchagne, routingKey)
	var err error
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "failed     to connect rabbingmq!")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

//路由模式發(fā)送信息
func (r *RabbitMQ) PublishRouting(message string) {
	//嘗試創(chuàng)建交換機(jī)，不存在創(chuàng)建
	err := r.channel.ExchangeDeclare(
		//交換機(jī)名稱(chēng)
		r.Exchange,
		//交換機(jī)類(lèi)型 廣播類(lèi)型
		"direct",
		//是否持久化
		true,
		//是否字段刪除
		false,
		//true表示這個(gè)exchange不可以被client用來(lái)推送消息，僅用來(lái)進(jìn)行exchange和exchange之間的綁定
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		nil,
	)
	r.failOnErr(err, "failed to declare an excha"+"nge")
	//發(fā)送信息
	err = r.channel.Publish(
		r.Exchange,
		//要設(shè)置
		r.Key,
		false,
		false,
		amqp.Publishing{
			//類(lèi)型
			ContentType: "text/plain",
			//消息
			Body: []byte(message),
		})
}

//路由模式接收信息
func (r *RabbitMQ) RecieveRouting() {
	//嘗試創(chuàng)建交換機(jī)，不存在創(chuàng)建
	err := r.channel.ExchangeDeclare(
		//交換機(jī)名稱(chēng)
		r.Exchange,
		//交換機(jī)類(lèi)型 廣播類(lèi)型
		"direct",
		//是否持久化
		true,
		//是否字段刪除
		false,
		//true表示這個(gè)exchange不可以被client用來(lái)推送消息，僅用來(lái)進(jìn)行exchange和exchange之間的綁定
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		nil,
	)
	r.failOnErr(err, "failed to declare an excha"+"nge")
	//2試探性創(chuàng)建隊(duì)列，創(chuàng)建隊(duì)列
	q, err := r.channel.QueueDeclare(
		"", //隨機(jī)生產(chǎn)隊(duì)列名稱(chēng)
		false,
		false,
		true,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	//綁定隊(duì)列到exchange中
	err = r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，這里的key要為空
		r.Key,
		r.Exchange,
		false,
		nil,
	)
	//消費(fèi)消息
	message, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)
	go func() {
		for d := range message {
			log.Printf("Received a message:%s,", d.Body)
		}
	}()
	fmt.Println("退出請(qǐng)按 Ctrl+C")
	<-forever
}

//簡(jiǎn)單模式Step:2、簡(jiǎn)單模式下生產(chǎn)代碼
func (r *RabbitMQ) PublishSimple(message string) {
	//1、申請(qǐng)隊(duì)列，如果隊(duì)列存在就跳過(guò)，不存在創(chuàng)建
	//優(yōu)點(diǎn)：保證隊(duì)列存在，消息能發(fā)送到隊(duì)列中
	_, err := r.channel.QueueDeclare(
		//隊(duì)列名稱(chēng)
		r.QueueName,
		//是否持久化
		false,
		//是否為自動(dòng)刪除 當(dāng)最后一個(gè)消費(fèi)者斷開(kāi)連接之后，是否把消息從隊(duì)列中刪除
		false,
		//是否具有排他性 true表示自己可見(jiàn) 其他用戶(hù)不能訪(fǎng)問(wèn)
		false,
		//是否阻塞 true表示要等待服務(wù)器的響應(yīng)
		false,
		//額外數(shù)據(jù)
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}

	//2.發(fā)送消息到隊(duì)列中
	r.channel.Publish(
		//默認(rèn)的Exchange交換機(jī)是default,類(lèi)型是direct直接類(lèi)型
		r.Exchange,
		//要賦值的隊(duì)列名稱(chēng)
		r.QueueName,
		//如果為true，根據(jù)exchange類(lèi)型和routkey規(guī)則，如果無(wú)法找到符合條件的隊(duì)列那么會(huì)把發(fā)送的消息返回給發(fā)送者
		false,
		//如果為true,當(dāng)exchange發(fā)送消息到隊(duì)列后發(fā)現(xiàn)隊(duì)列上沒(méi)有綁定消費(fèi)者，則會(huì)把消息還給發(fā)送者
		false,
		//消息
		amqp.Publishing{
			//類(lèi)型
			ContentType: "text/plain",
			//消息
			Body: []byte(message),
		})
}

func (r *RabbitMQ) ConsumeSimple() {
	//1、申請(qǐng)隊(duì)列，如果隊(duì)列存在就跳過(guò)，不存在創(chuàng)建
	//優(yōu)點(diǎn)：保證隊(duì)列存在，消息能發(fā)送到隊(duì)列中
	_, err := r.channel.QueueDeclare(
		//隊(duì)列名稱(chēng)
		r.QueueName,
		//是否持久化
		false,
		//是否為自動(dòng)刪除 當(dāng)最后一個(gè)消費(fèi)者斷開(kāi)連接之后，是否把消息從隊(duì)列中刪除
		false,
		//是否具有排他性
		false,
		//是否阻塞
		false,
		//額外數(shù)據(jù)
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	//接收消息
	msgs, err := r.channel.Consume(
		r.QueueName,
		//用來(lái)區(qū)分多個(gè)消費(fèi)者
		"",
		//是否自動(dòng)應(yīng)答
		true,
		//是否具有排他性
		false,
		//如果設(shè)置為true,表示不能同一個(gè)connection中發(fā)送的消息傳遞給這個(gè)connection中的消費(fèi)者
		false,
		//隊(duì)列是否阻塞
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	forever := make(chan bool)

	//啟用協(xié)程處理
	go func() {
		for d := range msgs {
			//實(shí)現(xiàn)我們要處理的邏輯函數(shù)
			log.Printf("Received a message:%s", d.Body)
			//fmt.Println(d.Body)
		}
	}()

	log.Printf("【*】warting for messages, To exit press CCTRAL+C")
	<-forever
}

func (r *RabbitMQ) ConsumeWorker(consumerName string) {
	//1、申請(qǐng)隊(duì)列，如果隊(duì)列存在就跳過(guò)，不存在創(chuàng)建
	//優(yōu)點(diǎn)：保證隊(duì)列存在，消息能發(fā)送到隊(duì)列中
	_, err := r.channel.QueueDeclare(
		//隊(duì)列名稱(chēng)
		r.QueueName,
		//是否持久化
		false,
		//是否為自動(dòng)刪除 當(dāng)最后一個(gè)消費(fèi)者斷開(kāi)連接之后，是否把消息從隊(duì)列中刪除
		false,
		//是否具有排他性
		false,
		//是否阻塞
		false,
		//額外數(shù)據(jù)
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	//接收消息
	msgs, err := r.channel.Consume(
		r.QueueName,
		//用來(lái)區(qū)分多個(gè)消費(fèi)者
		consumerName,
		//是否自動(dòng)應(yīng)答
		true,
		//是否具有排他性
		false,
		//如果設(shè)置為true,表示不能同一個(gè)connection中發(fā)送的消息傳遞給這個(gè)connection中的消費(fèi)者
		false,
		//隊(duì)列是否阻塞
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	forever := make(chan bool)

	//啟用協(xié)程處理
	go func() {
		for d := range msgs {
			//實(shí)現(xiàn)我們要處理的邏輯函數(shù)
			log.Printf("%s Received a message:%s", consumerName, d.Body)
			//fmt.Println(d.Body)
		}
	}()

	log.Printf("【*】warting for messages, To exit press CCTRAL+C")
	<-forever
}

func main() {
	//消費(fèi)者
	rabbitmq := RabbitMQ.NewRabbitMQSimple("ziyiSimple")
	rabbitmq.ConsumeSimple()
}

func main() {
	//Simple模式 生產(chǎn)者
	rabbitmq := RabbitMQ.NewRabbitMQSimple("ziyiSimple")
	for i := 0; i < 5; i++ {
		time.Sleep(time.Second * 2)
		rabbitmq.PublishSimple(fmt.Sprintf("%s %d", "hello", i))
	}
}

func main() {
	/*
		worker模式無(wú)非就是多個(gè)消費(fèi)者去同一個(gè)隊(duì)列中消費(fèi)消息
	*/
	//消費(fèi)者1
	rabbitmq1 := RabbitMQ.NewRabbitMQSimple("ziyiWorker")
	go rabbitmq1.ConsumeWorker("consumer1")
	//消費(fèi)者2
	rabbitmq2 := RabbitMQ.NewRabbitMQSimple("ziyiWorker")
	rabbitmq2.ConsumeWorker("consumer2")
}

func main() {
	//Worker模式 生產(chǎn)者
	rabbitmq := RabbitMQ.NewRabbitMQSimple("ziyiWorker")
	for i := 0; i < 100; i++ {
		//time.Sleep(time.Second * 2)
		rabbitmq.PublishSimple(fmt.Sprintf("%s %d", "hello", i))
	}
}

func main() {
	//消費(fèi)者
	rabbitmq := RabbitMQ.NewRabbitMQPubSub("" + "newProduct")
	rabbitmq.RecieveSub()
}

func main() {
	//訂閱模式發(fā)送者
	rabbitmq := RabbitMQ.NewRabbitMQPubSub("" + "newProduct")
	for i := 0; i <= 20; i++ {
		rabbitmq.PublishPub("訂閱模式生產(chǎn)第" + strconv.Itoa(i) + "條數(shù)據(jù)")
		fmt.Println(i)
		time.Sleep(1 * time.Second)
	}
}

func main() {
	//消費(fèi)者
	rabbitmq := RabbitMQ.NewRabbitMQRouting("exZi", "imooc_one")
	rabbitmq.RecieveRouting()
}

func main() {
	//路由模式生產(chǎn)者
	imoocOne := RabbitMQ.NewRabbitMQRouting("exZi", "imooc_one")
	imoocTwo := RabbitMQ.NewRabbitMQRouting("exZi", "imooc_two")

	for i := 0; i <= 10; i++ {
		imoocOne.PublishRouting("hello imooc one!" + strconv.Itoa(i))
		imoocTwo.PublishRouting("hello imooc two!" + strconv.Itoa(i))
		time.Sleep(1 * time.Second)
		fmt.Println(i)
	}
}

func main() {
	/*
		星號(hào)井號(hào)代表通配符
		星號(hào)代表多個(gè)單詞,井號(hào)代表一個(gè)單詞
		路由功能添加模糊匹配
		消息產(chǎn)生者產(chǎn)生消息,把消息交給交換機(jī)
		交換機(jī)根據(jù)key的規(guī)則模糊匹配到對(duì)應(yīng)的隊(duì)列,由隊(duì)列的監(jiān)聽(tīng)消費(fèi)者接收消息消費(fèi)
	*/
	//Topic消費(fèi)者
	//rabbitmq := RabbitMQ.NewRabbitMQTopic("exImoocTopic", "#") //匹配所有的key:topic88和topic99
	rabbitmq := RabbitMQ.NewRabbitMQTopic("exImoocTopic", "imooc.topic88.three") //只匹配topic88的
	rabbitmq.RecieveTopic()
}

func main() {
	//Topic模式生產(chǎn)者
	imoocOne := RabbitMQ.NewRabbitMQTopic("exImoocTopic", "imooc.topic88.three")
	imoocTwo := RabbitMQ.NewRabbitMQTopic("exImoocTopic", "imooc.topic99.four")

	for i := 0; i <= 10; i++ {
		imoocOne.PublishTopic("hello imooc topic three!" + strconv.Itoa(i))
		imoocTwo.PublishTopic("hello imooc topic four!" + strconv.Itoa(i))
		time.Sleep(1 * time.Second)
		fmt.Println(i)
	}
}

 vim docker-compose.yml

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:6.2.0
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  kafka:
    image: confluentinc/cp-kafka:6.2.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      #KAFKA_ADVERTISED_LISTENERS后面改為自己本地宿主機(jī)的ip，例如我本地mac的ip為192.168.0.101
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.0.101:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    depends_on:
      - zookeeper

# 進(jìn)入到docker-compose.yml所在目錄，執(zhí)行下面命令
docker-compose up -d
# 查看部署結(jié)果,狀態(tài)為up表明部署成功
docker-compose ps 

# 1. 創(chuàng)建對(duì)應(yīng)topic
docker-compose exec kafka kafka-topics --create --topic test-topic --partitions 1 --replication-factor 1 --bootstrap-server 192.168.0.101:9092

# 2. 查看topic列表
docker-compose exec kafka kafka-topics --list --zookeeper zookeeper:2181

package main

import (
	"fmt"

	"github.com/IBM/sarama"
)

// 基于sarama第三方庫(kù)開(kāi)發(fā)的kafka client

func main() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 發(fā)送完數(shù)據(jù)需要leader和follow都確認(rèn)
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新選出一個(gè)partition
	config.Producer.Return.Successes = true                   // 成功交付的消息將在success channel返回

	// 構(gòu)造一個(gè)消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = "web_log"
	msg.Value = sarama.StringEncoder("this is a test log")
	// 連接kafka
	client, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	defer client.Close()
	// 發(fā)送消息
	pid, offset, err := client.SendMessage(msg)
	if err != nil {
		fmt.Println("send msg failed, err:", err)
		return
	}
	fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

package main

import (
	"fmt"

	"github.com/IBM/sarama"
)

// kafka consumer

func main() {
	consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
	if err != nil {
		fmt.Printf("fail to start consumer, err:%v\n", err)
		return
	}
	partitionList, err := consumer.Partitions("web_log") // 根據(jù)topic取到所有的分區(qū)
	if err != nil {
		fmt.Printf("fail to get list of partition:err%v\n", err)
		return
	}
	fmt.Println(partitionList)
	for partition := range partitionList { // 遍歷所有的分區(qū)
		// 針對(duì)每個(gè)分區(qū)創(chuàng)建一個(gè)對(duì)應(yīng)的分區(qū)消費(fèi)者
		pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
			return
		}
		defer pc.AsyncClose()
		// 異步從每個(gè)分區(qū)消費(fèi)信息
		go func(sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
			}
		}(pc)
	}
	//演示時(shí)使用
	select {}
}