GoLang RabbitMQ实现六种工作模式示例

2024-11-27 16:05:55

六种工作模式介绍

1.简单（Simple）模式

P：生产者，也就是要发送消息的程序。

C：消费者：消息的接收者，会一直等待消息到来。

queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

简单模式就是单发单收，消息的消费者监听消息队列，如果队列中有消息，就消费掉，消息被拿走后，自动从队列中删除。

2.工作队列（Work Queue）模式

这种模式就是多个消费者消费同一个队列中的消息，既然消费者多了那么就出现了消息分配的问题，所以对应着两种分配策略：

公平分发：每个消费者接收消息的概率是相等的，消息队列会循环依次给每个消费者发送消息，这种是默认的策略。
公平派遣：保证消费者在消费完某个消息，并发送确认信息后，消息队列才会向它推送新的消息，在此之间若是有新的消息，将会被推送到其它消费者，若所有的消费者都在消费消息，那么就会等待。

3.发布/订阅（Pub/Sub）模式

在这种模型中，多了一个 Exchange 角色，而且过程略有变化：

P：生产者,也就是要发送消息的程序，但是不再发送到队列中，而是发给X (交换机)。

C：消费者，消息的接收者，会一直等待消息到来。

Queue：消息队列，接收消息、缓存消息。

Exchange：交换机(X) ，一方面，接收生产者发送的消息。另一方面，如何处理消息，递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。 Exchange有以下4种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列。
Direct：全值匹配，把消息交给符合指定routing key的队列。
Topic：通配符，与Direct类型类似，但Direct类型要求routing key完全相等，而Topic类型是对routing key进行模糊匹配，比Direct灵活。
Headers：根据Message的一些头部信息来分发过滤Message，用的比较少。

注意：Exchange负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失。

4.路由（Routing）模式

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Direct类型。在这种模式下：

Exchange 不再把消息交给每一个绑定的队列，而是根据消息的routing key进行判断，只有队列的

routing key与消息的routing key完全一致，才会接收到消息。

P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列。

C1：消费者，其所在队列指定了需要routing key为error的消息。

C2：消费者，其所在队列指定了需要routing key为 info、error、warning 的消息。

5.通配符（Tpoic）模式

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Topic类型。在这种模式下：

队列的routing key与消息的routing key符合匹配规则，就可以接收到消息，有两种规则：

*：可以（只能）匹配一个单词。

#：可以匹配多个单词（或者零个）。

所以图中，routing key为a.orange.b的消息就会被转发到Q1，而routing key为Lazy.a.b.c的消息就会被发送到Q2。

Go语言的实现

安装操作库

安装API库

Go可以使用streadway/amqp库来操作rabbit，使用以下命令来安装：

go get github.com/streadway/amqp

封装rabbitmq

接下来我们对streadway/amqp库的内容进行一个二次封装，封装为一个rabbitmq.go文件：

package rabbitmq
import (
	"encoding/json"
	"github.com/streadway/amqp"
	"log"
)
// RabbitMQ RabbitMQ结构
type RabbitMQ struct {
	channel  *amqp.Channel
	Name     string
	exchange string
}
// Connect 连接服务器
func Connect(s string) *RabbitMQ {
	//连接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "连接Rabbitmq服务器失败！")
	ch, e := conn.Channel()
	failOnError(e, "无法打开频道！")
	mq := new(RabbitMQ)
	mq.channel = ch
	return mq
}
// New 初始化消息队列
//第一个参数：rabbitmq服务器的链接，第二个参数：队列名字
func New(s string, name string) *RabbitMQ {
	//连接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "连接Rabbitmq服务器失败！")
	ch, e := conn.Channel()
	failOnError(e, "无法打开频道！")
	q, e := ch.QueueDeclare(
		name,  //队列名
		false, //是否开启持久化
		true,  //不使用时删除
		false, //排他
		false, //不等待
		nil,   //参数
	)
	failOnError(e, "初始化消息队列失败！")
	mq := new(RabbitMQ)
	mq.channel = ch
	mq.Name = q.Name
	return mq
}
// QueueDeclare 声明queue
func (q *RabbitMQ) QueueDeclare(queue string) {
	_, e := q.channel.QueueDeclare(queue, false, true, false, false, nil)
	failOnError(e, "声明queue失败！")
}
// QueueDelete 删除queue
func (q *RabbitMQ) QueueDelete(queue string) {
	_, e := q.channel.QueueDelete(queue, false, true, false)
	failOnError(e, "删除queue失败！")
}
// Qos 配置queue参数
func (q *RabbitMQ) Qos() {
	e := q.channel.Qos(1, 0, false)
	failOnError(e, "无法设置QoS")
}
// NewExchange 初始化交换机
//第一个参数：rabbitmq服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型
func NewExchange(s string, name string, typename string) {
	//连接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "连接Rabbitmq服务器失败！")
	ch, e := conn.Channel()
	failOnError(e, "无法打开频道！")
	e = ch.ExchangeDeclare(
		name,     // name
		typename, // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(e, "初始化交换机失败！")
}
// ExchangeDelete 删除交换机
func (q *RabbitMQ) ExchangeDelete(exchange string) {
	e := q.channel.ExchangeDelete(exchange, false, true)
	failOnError(e, "删除交换机失败！")
}
// Bind 绑定消息队列到exchange
func (q *RabbitMQ) Bind(exchange string, key string) {
	e := q.channel.QueueBind(
		q.Name,
		key,
		exchange,
		false,
		nil,
	)
	failOnError(e, "绑定队列失败！")
	q.exchange = exchange
}
// Send 向消息队列发送消息
//Send方法可以往某个消息队列发送消息
func (q *RabbitMQ) Send(queue string, body interface{}) {
	str, e := json.Marshal(body)
	failOnError(e, "消息序列化失败！")
	e = q.channel.Publish(
		"",    //交换
		queue, //路由键
		false, //必填
		false, //立即
		amqp.Publishing{
			ReplyTo: q.Name,
			Body:    []byte(str),
		})
	msg := "向队列:" + q.Name + "发送消息失败！"
	failOnError(e, msg)
}
// Publish 向exchange发送消息
//Publish方法可以往某个exchange发送消息
func (q *RabbitMQ) Publish(exchange string, body interface{}, key string) {
	str, e := json.Marshal(body)
	failOnError(e, "消息序列化失败！")
	e = q.channel.Publish(
		exchange,
		key,
		false,
		false,
		amqp.Publishing{ReplyTo: q.Name,
			Body: []byte(str)},
	)
	failOnError(e, "向交换机发送消息失败！")
}
// Consume 接收某个消息队列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
	c, e := q.channel.Consume(
		q.Name, //指定从哪个队列中接收消息
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	failOnError(e, "接收消息失败！")
	return c
}
// Close 关闭队列连接
func (q *RabbitMQ) Close() {
	q.channel.Close()
}
//错误处理函数
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

简单（Simple）模式

生产者

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
	i := 0
	for {
		// 每隔2s发送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("simple", " simple message: "+strconv.Itoa(i))
		i = i + 1
	}
}

消费者

func main() {
	consumer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
	//接收消息时，指定
	messages := consumer.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

运行结果：

2022/11/05 18:54:47 Received a message: " simple message: 0"
2022/11/05 18:54:52 Received a message: " simple message: 1"
2022/11/05 18:54:57 Received a message: " simple message: 2"

工作队列（Work Queue）模式

公平分发模式：

公平分发模式采用的是轮询机制，它会将数个任务按顺序平均分发给消费者。

生产者

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	i := 0
	for {
		// 每隔2s发送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("worker", " worker message: "+strconv.Itoa(i))
		i = i + 1
	}
}

消费者1

func main() {
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	//接收消息
	messages := consumer1.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

消费者2

func main() {
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	//接收消息
	messages := consumer2.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

运行结果：

# 消费者1
2022/11/05 19:45:03 Received a message: " worker message: 0"
2022/11/05 19:45:07 Received a message: " worker message: 2"
2022/11/05 19:45:11 Received a message: " worker message: 4"

# 消费者2
2022/11/05 19:45:05 Received a message: " worker message: 1"
2022/11/05 19:45:09 Received a message: " worker message: 3"
2022/11/05 19:45:13 Received a message: " worker message: 5"

可以发现，公平模式下，偶数消息都被发送给了消费者1，而奇数消息都被发送给了消费者2。

公平派遣模式：

有时候，如果消息之间的复杂度不同，那么不同消费者消费消息所用的时间会不同。这个时候如果使用公平派发模式，可能会造成某一个消费者需要消费的消息积压过多。可以采用公平派遣模式：

公平派遣模式下发送端与公平分发相同，消费者端只需要加一段配置代码，我们可以将预取计数设置为1。这告诉RabbitMQ一次不要给消费者一个以上的消息。换句话说，在处理并确认上一条消息之前，不要将新消息发送给消费者。而是将其分派给不忙的下一个消费者。

关于消息的确认：

为了确保消息永不丢失，RabbitMQ支持消息确认。消费者发送回一个确认（acknowledgement），以告知RabbitMQ已经接收，处理了特定的消息，并且RabbitMQ可以自由删除它。

我们之前的代码中，RabbitMQ一旦向消费者传递了一条消息，便立即将其标记为删除（调用Consumer的第三个参数是autoAck，表示是否自动回复）。在这种情况下，如果你终止一个消费者那么你就可能会丢失这个任务，我们还将丢失所有已经交付给这个消费者的尚未消费的消息。如果一个消费者意外宕机了，那么我们希望将任务交付给其他消费者来消费者。

所以一旦向消费者传递了一条消息，就不能马上将其标记为删除，而是要手动确认。我们需要在创建消费者的时候将autoAck参数标记为false：

// Consume 接收某个消息队列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
	c, e := q.channel.Consume(
		q.Name, //指定从哪个队列中接收消息
		"",
		false, // 不自动确认消息
		false,
		false,
		false,
		nil,
	)
	failOnError(e, "接收消息失败！")
	return c
}

然后每消费完一条消息需要调用Ack(false)函数手动回复。

生产者

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	i := 0
	for {
		// 每隔2s发送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("worker", " worker message: "+strconv.Itoa(i))
		i = i + 1
	}
}

消费端限流：

实现公平派遣模式我们需要设置消费者端一次只能消费一条消息，之前我们已经进行了封装，直接在消费者端调用即可：

// Qos 配置queue参数
func (q *RabbitMQ) Qos() {
	e := q.channel.Qos(1, 0, false)
	failOnError(e, "无法设置QoS")
}

消费者1

func main() {
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	// 指定一次只消费一条消息，直到消费完才重新接收
	consumer1.Qos()
	//接收消息
	messages := consumer1.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用10s
			time.Sleep(time.Second * 10)
			// 手动回复
			ch.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	// 指定一次只消费一条消息，直到消费完才重新接收
	consumer2.Qos()
	//接收消息
	messages := consumer2.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消费消息要用2s
			time.Sleep(time.Second * 2)
			// 手动回复
			ch.Ack(false)
		}
	}()
	select {}
}

运行结果：

# 消费者1
2022/11/05 20:31:26 Received a message: " worker message: 0"
2022/11/05 20:31:36 Received a message: " worker message: 5"

# 消费者2
2022/11/05 20:31:28 Received a message: " worker message: 1"
2022/11/05 20:31:30 Received a message: " worker message: 2"
2022/11/05 20:31:32 Received a message: " worker message: 3"
2022/11/05 20:31:34 Received a message: " worker message: 4"
2022/11/05 20:31:38 Received a message: " worker message: 6"
2022/11/05 20:31:40 Received a message: " worker message: 7"

发布/订阅（Pub/Sub）模式

生产者

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange1", "fanout")
	i := 0
	for {
		time.Sleep(time.Second)
		// fanout模式下不用routing key
		producer.Publish("exchange1", "pubsub message: "+strconv.Itoa(i), "")
		i = i + 1
	}
}

消费者1

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 队列绑定到exchange
	consumer1.Bind("exchange1", "")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 队列绑定到exchange
	consumer2.Bind("exchange1", "")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

运行结果：

# 消费者1
2022/11/05 22:32:19 Consumer1 received a message: "pubsub message: 0"
2022/11/05 22:32:20 Consumer1 received a message: "pubsub message: 1"
2022/11/05 22:32:21 Consumer1 received a message: "pubsub message: 2"
2022/11/05 22:32:22 Consumer1 received a message: "pubsub message: 3"
2022/11/05 22:32:23 Consumer1 received a message: "pubsub message: 4"
2022/11/05 22:32:24 Consumer1 received a message: "pubsub message: 5"

# 消费者2
2022/11/05 22:32:19 Consumer2 received a message: "pubsub message: 0"
2022/11/05 22:32:20 Consumer2 received a message: "pubsub message: 1"
2022/11/05 22:32:21 Consumer2 received a message: "pubsub message: 2"
2022/11/05 22:32:22 Consumer2 received a message: "pubsub message: 3"
2022/11/05 22:32:23 Consumer2 received a message: "pubsub message: 4"
2022/11/05 22:32:24 Consumer2 received a message: "pubsub message: 5"

路由（Routing）模式

生产者

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	// 指定为direct类型
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange", "direct")
	i := 0
	for {
		time.Sleep(time.Second)
		// 如果是奇数，就发key1
		// 如果是偶数，就发key2
		if i%2 != 0 {
			producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key1")
		} else {
			producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key2")
		}
		i = i + 1
	}
}

消费者1

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 队列绑定到exchange
	consumer1.Bind("exchange", "key1")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 队列绑定到exchange
	consumer2.Bind("exchange", "key2")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

运行结果：

# 消费者1
2022/11/05 22:51:10 Consumer1 received a message: "routing message: 1"
2022/11/05 22:51:12 Consumer1 received a message: "routing message: 3"
2022/11/05 22:51:14 Consumer1 received a message: "routing message: 5"
2022/11/05 22:51:16 Consumer1 received a message: "routing message: 7"

# 消费者2
2022/11/05 22:51:11 Consumer2 received a message: "routing message: 0"
2022/11/05 22:51:13 Consumer2 received a message: "routing message: 2"
2022/11/05 22:51:15 Consumer2 received a message: "routing message: 4"
2022/11/05 22:51:17 Consumer2 received a message: "routing message: 6"

通配符（Tpoic）模式

生产者

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	// 指定为topic类型
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange2", "topic")
	var i int
	for {
		time.Sleep(time.Second)
		if i%2 != 0 {
			producer.Publish("exchange2", "topic message: "+strconv.Itoa(i), "a.test.b.c")
		} else {
			producer.Publish("exchange2", "topic message: "+strconv.Itoa(i), "a.test.b")
		}
		i++
	}
}

消费者1

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 队列绑定到exchange
	consumer1.Bind("exchange2", "*.test.*")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

消费者2

func main() {
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 队列绑定到exchange
	consumer2.Bind("exchange2", "#.test.#")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

运行结果：

# 消费者1
2022/11/05 23:09:53 Consumer1 received a message: "topic message: 0"
2022/11/05 23:09:55 Consumer1 received a message: "topic message: 2"
2022/11/05 23:09:57 Consumer1 received a message: "topic message: 4"
2022/11/05 23:09:59 Consumer1 received a message: "topic message: 6"

# 消费者2
2022/11/05 23:09:53 Consumer2 received a message: "topic message: 0"
2022/11/05 23:09:54 Consumer2 received a message: "topic message: 1"
2022/11/05 23:09:55 Consumer2 received a message: "topic message: 2"
2022/11/05 23:09:56 Consumer2 received a message: "topic message: 3"
2022/11/05 23:09:57 Consumer2 received a message: "topic message: 4"
2022/11/05 23:09:58 Consumer2 received a message: "topic message: 5"
2022/11/05 23:09:59 Consumer2 received a message: "topic message: 6"

到此这篇关于GoLang RabbitMQ实现六种工作模式示例的文章就介绍到这了,更多相关GoLang RabbitMQ工作模式内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！

GoLang RabbitMQ TTL与死信队列以及延迟队列详细讲解

目录 TTL 死信队列延迟队列 Go实现延迟队列 TTL TTL 全称 Time To Live(存活时间/过期时间).当消息到达存活时间后,还没有被消费,就会被自动清除.RabbitMQ可以设置两种过期时间: 对消息设置过期时间. 对整个队列(Queue)设置过期时间. 如何设置设置队列过期时间使用参数:x-message-ttl,单位:ms(毫秒),会对整个队列消息统一过期. 设置消息过期时间使用参数:expiration,单位:ms(毫秒),当该消息在队列头部时(消费时),会单独判断这
golang gin 监听rabbitmq队列无限消费的案例代码

golang gin 监听rabbitmq队列无限消费连接rabbitmq package database import ( "github.com/streadway/amqp" "log" "reflect" "yy-data-processing/common/config" ) var RabbitConn *amqp.Connection var RabbitChannel *amqp.Channel func
关于golang监听rabbitmq消息队列任务断线自动重连接的问题

golang监听消息队列rabbitmq任务脚本,当rabbimq消息队列断开连接后自动重试,重新唤起协程执行任务需求背景: goalng常驻内存任务脚本监听rbmq执行任务任务脚本由supervisor来管理当rabbitmq长时间断开连接会出现如下图进程处于fatal状态假如因为不可抗拒因素,rabbitmq服务器内存满了或者其它原因导致rabbitmq消息队列服务停止了如果是短时间的停止重启,supervisor是可以即时唤醒该程序.如果服务器长时间没有恢复正常运行,程序就会出
Golang rabbitMQ生产者消费者实现示例

目录消费者生产者消费者 package main import ( "fmt" "github.com/streadway/amqp" ) func failOnError(err error, msg string) { if err != nil { fmt.Println("%s: %s", msg, err) } } // 只能在安装 rabbitmq 的服务器上操作 func main() { conn, err := amqp.
GoLang RabbitMQ实现六种工作模式示例

目录六种工作模式介绍 1.简单(Simple)模式 2.工作队列(Work Queue)模式 3.发布/订阅(Pub/Sub)模式 4.路由(Routing)模式 5.通配符(Tpoic)模式 Go语言的实现安装操作库简单(Simple)模式工作队列(Work Queue)模式发布/订阅(Pub/Sub)模式路由(Routing)模式通配符(Tpoic)模式六种工作模式介绍 1.简单(Simple)模式 P:生产者,也就是要发送消息的程序. C:消费者:消息的接收者,会一直等待消
SpringBoot整合RabbitMQ实现六种工作模式的示例

目录前提概念生产者队列消费者 SpringBoot整合RabbitMQ基本配置添加maven依赖 1. 简单(simple)模式 2. 工作模式生产消息: 3. 发布订阅模式特点创建队列.交换机以及绑定: 4. 路由模式特点创建队列.交换机以及绑定: 5. 主题模式特点创建交换机和队列: 6. RPC模式特点消费端添加返回值: 交换机类型 Direct Exchange(直连) Fanout Exchange(扇形) Topic Exchange(主题) 总结源码示例
Golang WorkerPool线程池并发模式示例详解

目录正文处理CVS文件记录获取测试数据线程池耗时差异正文 Worker Pools 线程池是一种并发模式.该模式中维护了固定数量的多个工作器,这些工作器等待着管理者分配可并发执行的任务.该模式避免了短时间任务创建和销毁线程的代价. 在 golang 中,我们使用 goroutine 和 channel 来构建这种模式.工作器 worker 由一个 goroutine 定义,该 goroutine 通过 channel 获取数据. 处理CVS文件记录接下来让我们通过一个例子,来进一步理
SpringBoot整合RabbitMQ的5种模式实战

目录一.环境准备二.简单模式三.工作队列模式四.广播模式(Fanout) 五.直连模式(Direct) 六.通配符模式(Topic) 一.环境准备 1.pom依赖  <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version&g
python操作RabbitMq的三种工作模式

一.简介: RabbitMq 是实现了高级消息队列协议(AMQP)的开源消息代理中间件.消息队列是一种应用程序对应用程序的通行方式,应用程序通过写消息,将消息传递于队列,由另一应用程序读取完成通信.而作为中间件的 RabbitMq 无疑是目前最流行的消息队列之一. RabbitMq 应用场景广泛: 系统的高可用:日常生活当中各种商城秒杀,高流量,高并发的场景.当服务器接收到如此大量请求处理业务时,有宕机的风险.某些业务可能极其复杂,但这部分不是高时效性,不需要立即反馈给用户,我们可以将这部
Golang设计模式中抽象工厂模式详细讲解

目录抽象工厂模式概念示例抽象工厂模式抽象工厂模式是一种创建型设计模式, 它能创建一系列相关的对象, 而无需指定其具体类. 抽象工厂定义了用于创建不同产品的接口, 但将实际的创建工作留给了具体工厂类. 每个工厂类型都对应一个特定的产品变体. 在创建产品时, 客户端代码调用的是工厂对象的构建方法, 而不是直接调用构造函数 (new操作符). 由于一个工厂对应一种产品变体, 因此它创建的所有产品都可相互兼容. 客户端代码仅通过其抽象接口与工厂和产品进行交互. 该接口允许同一客户端代码与不同产品
Spring Boot整合RabbitMQ实例（Topic模式）

1.Topic交换器介绍 Topic Exchange 转发消息主要是根据通配符. 在这种交换机下,队列和交换机的绑定会定义一种路由模式,那么,通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息. 在这种交换机模式下: 路由键必须是一串字符,用句号(.) 隔开,比如说 agreements.us,或者 agreements.eu.stockholm 等. 路由模式必须包含一个星号(*),主要用于匹配路由键指定位置的一个单词,比如说,一个路由模式是这样子:agreements..b.*
Java对称加密工作模式原理详解

这篇文章主要介绍了Java对称加密工作模式原理详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下对称加密又分为分组加密和序列密码. 分组密码,也叫块加密(block cyphers),一次加密明文中的一个块.是将明文按一定的位长分组,明文组经过加密运算得到密文组,密文组经过解密运算(加密运算的逆运算),还原成明文组. 序列密码,也叫流加密(stream cyphers),一次加密明文中的一个位.是指利用少量的密钥(制乱元素)通过某种复杂的运算
详解Python 实现 ZeroMQ 的三种基本工作模式

简介引用官方说法:ZMQ(以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层,像框架一样的一个 socket library,他使得 Socket 编程更加简单.简洁和性能更高. 是一个消息处理队列库,可在多个线程.内核和主机盒之间弹性伸缩. ZMQ 的明确目标是"成为标准网络协议栈的一部分,之后进入 Linux 内核".现在还未看到它们的成功.但是,它无疑是极具前景的.并且是人们更加需要的"传统" BSD 套接字之上的一层封装.ZMQ 让编写高性能网络应