RabbitMQ核心概念及AMQP协议

2023-12-16 18:32
文章标签 协议 概念 核心 rabbitmq amqp

本文主要是介绍RabbitMQ核心概念及AMQP协议，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

RabbitMQ核心概念及AMQP协议

前言

版本说明

rabbitmq=3.8.4

相关链接

  • RabbitMQ 官网文档：https://www.rabbitmq.com/getstarted.html
  • SpringBoot 整合 RabbitMQ maven 官网地址：https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

AMQP

AMQP（Advanced Message Queuing Protocol） ：高级消息队列协议，是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

现代特征特点：多信道、协商式、异步、安全、可扩展、中立、高效。

AMQP 协议模型

在这里插入图片描述

AMQP 核心概念

  • Server ：又称 Broker ，接收客户端的连接，实现 AMQP 实体服务；
  • Connection ：连接，应用程序与 Broker 的网络连接；
  • Channel ：网络信道，几乎所有的操作都在 Channel 中进行，Channel 是进行消息读写的通道。客户端可以建立多个 Channel ，每个 Channel 代表一个会话任务；
  • Message ：消息，服务器和应用程序之间传送的数据，由 Properties 和 Body 组成。Properties 可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body 则就是消息体的内容；
  • Virtual Host ：虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个 Virtual Host 里面可以有若干个 Exchange 和 Queue ，同一个 Virtual Host 里面不能有相同名称的 Exchange 或者 Queue ；
  • Exchange ：交换机，接收消息，根据路由键转发消息到绑定的队列中；
  • Binding ：Exchange 和 Queue 之间的虚拟连接，Binding 中可以包含 routing key ；
  • Routing Key ：一个路由规则，虚拟机可以用它来确定如何路由一个特定消息；
  • Queue ：也称为 Message Queue ，消息队列，保存消息并将它们转发给消费者；

RabbitMQ

RabbitMQ 优势

  • 与 SpringAMQP 完美的整合，API 丰富；
  • 集群模式丰富，表达式配置，HA 模式，镜像队列模型；
  • 保证数据不丢失的前提做到高可靠性、高可用性；

RabbitMQ 高性能原因

  • Erlang 语言最初在与交换机领域的架构模式，这样使得 RabbitMQ 在 Broker 之间进行数据交互的性能是非常优秀的；
  • Erlang 语言有着和原生 Socket 一样的延迟；

RabbitMQ 整体架构

在这里插入图片描述

RabbitMQ 消息流转

在这里插入图片描述

RabbitMQ 角色说明

  • 超级管理员（Administrator）：可登录管理控制台，并且可以对用户，策略进行操作；
  • 监控者（Monitoring）：可登录管理控制台，同时可以查看节点的相关信息（进程数、内存使用情况、磁盘使用情况等）；
  • 策略制定者（Policymaker）：可登录管理控制台，同时可以对策略进行管理，但无法查看节点相关信息；
  • 普通管理员（Management）：仅可以登录管理控制台，无法看到节点相关信息，也无法对策略进行管理；
  • 其他：无法登录管理控制台，通常就是普通的生产者和消费者；

命令行与管控台操作

基础操作

# 关闭应用
rabbitmqctl stop_app
# 启动应用
rabbitmqctl start_app
# 节点状态
rabbitmqctl status# 添加用户
rabbitmqctl add_user username password
# 列出所有用户
rabbitmqctl list_users
# 删除用户
rabbitmqctl delete_user username
# 清除用户权限
rabbitmqctl clear_permissions -p vhostpath username
# 列出用户权限
rabbitmqctl list_user_permissions username
# 修改密码
rabbitmqctl change_password username password
# 设置用户权限
rabbitmqctl set_permissions -p vhostpath username ".*" ".*" ".*" # 创建虚拟主机
rabbitmqctl add_vhost vhostpath
# 列出所有的虚拟主机
rabbitmqctl list_vhosts
# 列出虚拟主机上所有权限
rabbitmqctl list_permissions -p vhostpath
# 删除虚拟主机
rabbitmqctl delete_vhost vhostpath# 查看所有队列信息
rabbitmqctl list_queues
# 清除队列里的消息
rabbitmqctl -p vhostpath purge_queue blue

高级操作

# 移除所有数据，要在 rabbitmqctl stop_app 之后使用 
rabbitmqctl reset
# 组成集群命令
rabbitmqctl join_cluster <clusternode> 
# 查看集群状态
rabbitmqctl cluster_status
# 修改集群节点的存储形式
rabbitmqctl change_cluster_node_type disc|ram
# 忘记节点（摘除节点）
rabbitmqctl forget_cluster_node [--offline]
# 修改节点名称
rabbitmqctl rename_cluster_node oldnode1 newnode1 oldnode2 newnode2...

RabbitMQ 工作模式

  • P：生产者
  • C：消费者
  • Queue：红色部分
  • X：exchange 交换机，深蓝色部分。交换机一方面，接收生产者发送的消息；另一方面，知道如何处理消息，例如如何递交给某个特定的队列、递交给所有的队列、或者是将消息丢弃。Exchange只负责转发消息，不具备存储消息的能力； Exchange 常见类型：
    1. Fanout ：广播，将消息交给所有绑定到交换机的队列；
    2. Direct：定向，把消息交给符合指定 routing key 规则的队列；
    3. Topic：通配符，把消息交给符合 routing pattern （路由模式）的队列；
    4. headers：首部交换机

SimpleQueue 简单模式

原理

The simplest thing that does something

在这里插入图片描述

生产者代码
package top.simba1949.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:52*/
public class Producer {public static final String SIMPLE_QUEUE_NAME = "simple_queue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.8.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 2.通过连接工厂创建连接Connection connection = connectionFactory.newConnection();// 3.通过 connection 创建 channelChannel channel = connection.createChannel();// 4.声明队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(SIMPLE_QUEUE_NAME, true, false, false, null);// 发送消息String msg = "Hello RabbitMQ";// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish("", SIMPLE_QUEUE_NAME, null, msg.getBytes());// 释放资源channel.close();connection.close();}
}
消费者代码
package top.simba1949.simple;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.8.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 2.通过连接工厂创建连接Connection connection = connectionFactory.newConnection();// 3.通过 connection 创建 channelChannel channel = connection.createChannel();// 4.声明一个队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(Producer.SIMPLE_QUEUE_NAME, true, false, false, null);// 5.监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));System.out.println("-----------------");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.SIMPLE_QUEUE_NAME, true, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}

WorkQueues 工作队列模式

原理

Distributing tasks among workers (the competing consumers pattern)

在这里插入图片描述

工具类
package top.simba1949;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/16 16:20*/
public class RabbitMQUtil {/*** 获取连接* @return* @throws IOException* @throws TimeoutException*/public static Connection getConnection() throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.8.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 2.通过连接工厂创建连接Connection connection = connectionFactory.newConnection();return connection;}/*** 关闭连接* @param channel* @param connection* @throws IOException* @throws TimeoutException*/public static void close(Channel channel, Connection connection) throws IOException, TimeoutException {channel.close();connection.close();}
}
生产者代码
package top.simba1949.work;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:52*/
public class Producer {public static final String QUEUE_NAME = "work_queue";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitMQUtil.getConnection();// 通过 connection 创建 channelChannel channel = connection.createChannel();// 声明队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 发送消息for (int i = 1; i <= 30; i++){String msg = "你好：小兔子！ work 模式————" + i;// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，简单模式可以传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());}// 释放资源RabbitMQUtil.close(channel, connection);}
}
消费者1代码
package top.simba1949.work;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明一个队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);// 监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============消费者1 start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("============消费者1 end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.QUEUE_NAME, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}
消费者2代码
package top.simba1949.work;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明一个队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);// 监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============消费者2 start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("============消费者2 end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.QUEUE_NAME, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}

Publish/Subscribe

原理

Sending messages to many consumers at once

在这里插入图片描述

工具类代码
package top.simba1949;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/16 16:20*/
public class RabbitMQUtil {/*** 获取连接* @return* @throws IOException* @throws TimeoutException*/public static Connection getConnection() throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.8.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 2.通过连接工厂创建连接Connection connection = connectionFactory.newConnection();return connection;}/*** 关闭连接* @param channel* @param connection* @throws IOException* @throws TimeoutException*/public static void close(Channel channel, Connection connection) throws IOException, TimeoutException {channel.close();connection.close();}
}
生产者代码
package top.simba1949.publish.subscribe;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:52*/
public class Producer {public static final String FANOUT_EXCHANGE = "fanout_exchange";public static final String FANOUT_QUEUE_1 = "fanout_queue_1";public static final String FANOUT_QUEUE_2 = "fanout_queue_2";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitMQUtil.getConnection();// 通过 connection 创建 channelChannel channel = connection.createChannel();// 声明交换机// 参数一：交换机名称// 参数二：交换机类型： DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);// 声明队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);// 队列绑定交换机// 参数一：队列名称// 参数二：交换机名称// 参数三：路由keychannel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHANGE, "");channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHANGE, "");// 发送消息for (int i = 1; i <= 30; i++){String msg = "你好：小兔子！ Publish/Subscribe模式(" + i;// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish(FANOUT_EXCHANGE, "", null, msg.getBytes());}// 释放资源RabbitMQUtil.close(channel, connection);}
}
消费者1订阅队列1代码
package top.simba1949.publish.subscribe;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明一个队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);// 监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============Publish/Subscribe 消费者1 start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("============Publish/Subscribe 消费者1 end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.FANOUT_QUEUE_1, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}
消费者2订阅队列2代码
package top.simba1949.publish.subscribe;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 4.声明一个队列channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);// 5.监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============Publish/Subscribe 消费者2 start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("============ Publish/Subscribe消费者2 end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.FANOUT_QUEUE_2, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}

Routing

原理

Receiving messages selectively

在这里插入图片描述

路由模式特点

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 Routing Key
  • 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 Routing Key
  • Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routing Key 与消息 Routing Key 完全一致，才会接收到消息；
工具类代码
package top.simba1949;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/16 16:20*/
public class RabbitMQUtil {/*** 获取连接* @return* @throws IOException* @throws TimeoutException*/public static Connection getConnection() throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.8.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 2.通过连接工厂创建连接Connection connection = connectionFactory.newConnection();return connection;}/*** 关闭连接* @param channel* @param connection* @throws IOException* @throws TimeoutException*/public static void close(Channel channel, Connection connection) throws IOException, TimeoutException {channel.close();connection.close();}
}
生产者代码
package top.simba1949.routing;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 路由模式的交换机类型为：direct* @author Theodore* @date 2020/6/15 17:52*/
public class Producer {public static final String DIRECT_EXCHANGE = "direct_exchange";public static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";public static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";public static final String ROUTING_KEY_INSERT = "insert";public static final String ROUTING_KEY_UPDATE = "update";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitMQUtil.getConnection();// 通过 connection 创建 channelChannel channel = connection.createChannel();// 声明交换机// 参数一：交换机名称// 参数二：交换机类型： DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);// 声明队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);// 队列绑定交换机// 参数一：队列名称// 参数二：交换机名称// 参数三：路由keychannel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, ROUTING_KEY_INSERT);channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHANGE, ROUTING_KEY_UPDATE);// 发送消息String msgInsert = "新增了商品，路由模式：routing key 为 insert";// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish(DIRECT_EXCHANGE, ROUTING_KEY_INSERT, null, msgInsert.getBytes());// 发送消息String msgUpdate = "新增了商品，路由模式：routing key 为 update";// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish(DIRECT_EXCHANGE, ROUTING_KEY_UPDATE, null, msgUpdate.getBytes());// 释放资源RabbitMQUtil.close(channel, connection);}
}
消费者 insert 代码
package top.simba1949.routing;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class ConsumerInsert {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明一个队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);// 监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============Routing insert start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("============Routing insert end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}
消费者 update 代码
package top.simba1949.routing;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class ConsumerUpdate {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 4.声明一个队列channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);// 5.监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============Routing update start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("===========Routing update end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}

Topics 通配符模式

原理

Receiving messages based on a pattern (topics)

在这里插入图片描述

Topic 类型与 Direct 相比，都是可以根据 Routing Key 把消息路由到不同的队列。只不过 Topic 类型的 Exchange 可以让队列在绑定 Routing Key 的时候可以使用通配符。

Routing Key 一般都是有一个或者多个单词组成，多个单词以“.” 分割，例如：item.insert

通配符规则

  • # ：匹配一个或者多个词；eg ：item.# 能够匹配 item.insert 或者 item.insert.abc ;
  • * ：匹配一个词；item.* 能够匹配 item.insert;
工具类代码
package top.simba1949;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/16 16:20*/
public class RabbitMQUtil {/*** 获取连接* @return* @throws IOException* @throws TimeoutException*/public static Connection getConnection() throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.8.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 2.通过连接工厂创建连接Connection connection = connectionFactory.newConnection();return connection;}/*** 关闭连接* @param channel* @param connection* @throws IOException* @throws TimeoutException*/public static void close(Channel channel, Connection connection) throws IOException, TimeoutException {channel.close();connection.close();}
}
生产者代码
package top.simba1949.topic;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 路由模式的交换机类型为：topic* @author Theodore* @date 2020/6/15 17:52*/
public class Producer {public static final String TOPIC_EXCHANGE = "topic_exchange";public static final String TOPIC_QUEUE_1 = "topic_queue_1";public static final String TOPIC_QUEUE_2 = "topic_queue_2";public static final String ROUTING_KEY_TOPIC_1 = "item.#";public static final String ROUTING_KEY_TOPIC_2 = "item.*";public static final String ROUTING_KEY_INSERT = "item.insert";public static final String ROUTING_KEY_UPDATE = "item.update";public static final String ROUTING_KEY_DELETE = "item.delete.abc";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitMQUtil.getConnection();// 通过 connection 创建 channelChannel channel = connection.createChannel();// 声明交换机// 参数一：交换机名称// 参数二：交换机类型： DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);// 声明队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(TOPIC_QUEUE_1, true, false, false, null);channel.queueDeclare(TOPIC_QUEUE_2, true, false, false, null);// 队列绑定交换机// 参数一：队列名称// 参数二：交换机名称// 参数三：路由keychannel.queueBind(TOPIC_QUEUE_1, TOPIC_EXCHANGE, ROUTING_KEY_TOPIC_1);channel.queueBind(TOPIC_QUEUE_2, TOPIC_EXCHANGE, ROUTING_KEY_TOPIC_2);// 发送消息String msgInsert = "新增了商品，Topic模式：routing key 为 item.insert";// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish(TOPIC_EXCHANGE, ROUTING_KEY_INSERT, null, msgInsert.getBytes());// 发送消息String msgUpdate = "更新了商品，Topic模式：routing key 为 item.update";// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish(TOPIC_EXCHANGE, ROUTING_KEY_UPDATE, null, msgUpdate.getBytes());// 发送消息String msgDelete = "删除了商品，Topic模式：routing key 为 item.delete.abc";// 参数一：交换机名称，如果没有指定则使用 Default Exchange;// 参数二：路由 key，如果是简单模式，可以直接传递队列名称// 参数三：消息其他属性// 参数四：消息内容channel.basicPublish(TOPIC_EXCHANGE, ROUTING_KEY_DELETE, null, msgDelete.getBytes());// 释放资源RabbitMQUtil.close(channel, connection);}
}
消费者1代码
package top.simba1949.topic;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明一个队列// 参数一：队列名称// 参数二：是否定义持久化队列// 参数三：是否独占本次连接// 参数四：是否在不使用的使用自动删除队列// 参数五：队列其他参数channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);// 监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("============end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.TOPIC_QUEUE_1, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}
消费者2代码
package top.simba1949.topic;import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import top.simba1949.RabbitMQUtil;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Theodore* @date 2020/6/15 17:49*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {// 创建连接Connection connection = RabbitMQUtil.getConnection();// 创建信道Channel channel = connection.createChannel();// 4.声明一个队列channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);// 5.监听消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** @param consumerTag 消费标签，在 channel.basicConsume 时候可以指定* @param envelope 消息包内容，可从中获取消息id、消息routingKey、交换机和重转标记（收到消息失败后是否需要重新发送该消息）* @param properties 消息属性* @param body 消息* @throws IOException*/@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {super.handleDelivery(consumerTag, envelope, properties, body);System.out.println("============start =============");System.out.println("路由 key ：" + envelope.getRoutingKey());System.out.println("交换机 ：" + envelope.getExchange());System.out.println("消息id ：" + envelope.getDeliveryTag());System.out.println("接收到的消息 ：" + new String(body, "UTF-8"));Thread.sleep(1000L);// 客户端自己确认消息收到ackchannel.basicAck(envelope.getDeliveryTag(), false);System.out.println("===========end =============");}};// 监听消息// 参数一：队列名称// 参数二：是否自动确认，设置为true表示消息接收到自动向mq回复接收到了，mq接收到消息之后会删除消息，设置为false则需要手动确认channel.basicConsume(Producer.TOPIC_QUEUE_2, false, defaultConsumer);// 不关闭资源，会一直监听// channel.close();// connection.close();}
}

RabbitMQ 高级

过期时间TTL

过期时间TTL表示可以对消息设置预期的时间，在这个时间内都可以被消费者接受获取；过了之后消息将自动被删除。

RabbitMQ 可以对消息和队列设置 TTL。

  1. 通过队列属性设置 TTL，队列中所有消息都有相同的过期时间；
  2. 对消息进行单独设置，每条消息 TTL 可以不同；

如果上述两种方法同时使用，则消息的过期时间以两者之间 TTL 较小的那个数值为准。消息在队列的生存时间一旦超过设置的 TTL 值，就称为 dead message 被投递到死信队列，消费者将无法接收到消息。

死信队列

DLX，全称为 Dead-Letter-Exchange ，可以称之为死信交换机。当消息在一个队列中变成死信之后，它能被重新发送到另一个交换机中，这个交换机就是 DLX，绑定 DLX 的队列就称之为死信队列；

消息变成死信，可能是由于以下的原因：

  • 消息被拒绝
  • 消息过期
  • 队列达到最大长度

DLX 也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，实际上就是设置某一个队列的属性。当这个队列中存在死信时，RabbitMQ 就会自动的将这个消息重新发布到设置的 DLX 上去，进而被路由到另一个队列，即死信队列。

要想使用死信队列，只需要在定义队列的时候设置队列参数 x-dead-letter-exchange 指定交换机即可；

在这里插入图片描述

延迟队列

延迟队列存储的对象是对应的延迟消息；所谓“延迟消息”是指当消息被发送以后，并不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费；

在 RabbitMQ 中延迟队列可以通过 过期时间和死信队列 来实现；

消息确认机制

确认并且保证消息被送达，提供两种方式：发布确认和事务；（两者不可同时使用）在 channel 为事务时，不可引入确认模式；通用 channel 为确认模式下，不可使用事务。

发布确认

有两种方式：消息发送成功确认和消息发送失败回调；

事务支持

场景：业务处理伴随着消息的发送，业务处理失败（事务回滚）后要求消息不发送。RabbitMQ 使用调用者的外部事务，通常是首选，因为它是非入侵的（低耦合）；

<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<!--confirm-callback="confirmCallback"表示消息发送失败回调-->
<!--return-callback="returnCallback"表示消息失败回调，同时需要配置mandatory="true"，否则消息则丢失-->
<!--channel-transacted="true"表示支持事务操作-->
<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"confirm-callback="confirmCallback"return-callback="returnCallback"mandatory="true"channel-transacted="true"/>
<!--平台事务管理器-->
<bean id="rabbitTransactionManager" class="org.springframework.amqp.rabbit.transaction.RabbitTransactionManager"><property name="connectionFactory" ref="connectionFactory"/>
</bean>

消息追踪

消息中心的消息追踪需要使用 Trace 实现，Trace 是 RabbitMQ 用于记录每次发送的消息，方便使用 RabbitMQ 的开发者调试、排错。可通过插件形式提供可视化界面。Trace 启动后会自动创建系统 Exchange：amq.rabbitmq.trace，每一个队列会自动绑定该 Exchange ，绑定后发送到该队列的消息会记录到 Trace 日志。

命令集描述
rabbitmq-plugins list查看插件列表
rabbitmq-plugins enable rabbitmq_tracingrabbitmq启动trace插件
rabbitmqctl trace_on打开 trace 开关
rabbitmqctl trace_on vhost-path打开 trace 的开关，追踪 vhost-path 日志
rabbitmqctl trace_off关闭 trace 的开关
rabbitmqctl-plugins disable rabbitmq_tracingrabbitmq 关闭 trace 插件
rabbitmqctl set_user_tags 用户名 administrator只有 administrator 的角色才能查看日志界面

安装插件并开启 trace_on 之后，会发现多了个 exchange : amq.rabbitmq.trace，类型为 topic 。

RabbitMQ 高可用集群

  1. 主备模式：用来实现 RabbitMQ 的高可用集群，一般是在并发和数据不是特别多的时候使用，当主节点挂点以后会从备份节点中选择一个节点出来作为主节点对外提供服务；消费者通过 HAProxy 访问；HAProxy 可以切换主备节点；
  2. 远程模式： 主要用来实现双活，简称为 “Shovel模式” ，就是把消息复制到不同的数据中心，让两个跨地域的集群互联；
  3. 镜像队列模式： 也别称为 Mirror 队列，主要是用来保证 MQ 消息的可靠性，通过消息复制的方式能够保证消息 100% 不丢失，同时该集群模式也是企业中使用最多的模式；
  4. 多活模式： 主要用来实现异地数据复制，Shovel 模式其实也可以实现，但是Shovel配置比较繁琐，而且还受版本限制；使用多活模式需要借助 federation 插件来实现集群与集群之间或者节点与节点之间的消息复制。

列会自动绑定该 Exchange ，绑定后发送到该队列的消息会记录到 Trace 日志。

命令集描述
rabbitmq-plugins list查看插件列表
rabbitmq-plugins enable rabbitmq_tracingrabbitmq启动trace插件
rabbitmqctl trace_on打开 trace 开关
rabbitmqctl trace_on vhost-path打开 trace 的开关，追踪 vhost-path 日志
rabbitmqctl trace_off关闭 trace 的开关
rabbitmqctl-plugins disable rabbitmq_tracingrabbitmq 关闭 trace 插件
rabbitmqctl set_user_tags 用户名 administrator只有 administrator 的角色才能查看日志界面

安装插件并开启 trace_on 之后，会发现多了个 exchange : amq.rabbitmq.trace，类型为 topic 。

RabbitMQ 高可用集群

  1. 主备模式：用来实现 RabbitMQ 的高可用集群，一般是在并发和数据不是特别多的时候使用，当主节点挂点以后会从备份节点中选择一个节点出来作为主节点对外提供服务；消费者通过 HAProxy 访问；HAProxy 可以切换主备节点；
  2. 远程模式： 主要用来实现双活，简称为 “Shovel模式” ，就是把消息复制到不同的数据中心，让两个跨地域的集群互联；
  3. 镜像队列模式： 也别称为 Mirror 队列，主要是用来保证 MQ 消息的可靠性，通过消息复制的方式能够保证消息 100% 不丢失，同时该集群模式也是企业中使用最多的模式；
  4. 多活模式： 主要用来实现异地数据复制，Shovel 模式其实也可以实现，但是Shovel配置比较繁琐，而且还受版本限制；使用多活模式需要借助 federation 插件来实现集群与集群之间或者节点与节点之间的消息复制。

这篇关于RabbitMQ核心概念及AMQP协议的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/501472。

相关文章

探索蓝牙协议的奥秘：用ESP32实现高质量蓝牙音频传输

探索蓝牙协议的奥秘：用ESP32实现高质量蓝牙音频传输

蓝牙（Bluetooth）是一种短距离无线通信技术，广泛应用于各种电子设备之间的数据传输。自1994年由爱立信公司首次提出以来，蓝牙技术已经经历了多个版本的更新和改进。本文将详细介绍蓝牙协议，并通过一个具体的项目——使用ESP32实现蓝牙音频传输，来展示蓝牙协议的实际应用及其优点。 蓝牙协议概述 蓝牙协议栈 蓝牙协议栈是蓝牙技术的核心，定义了蓝牙设备之间如何进行通信。蓝牙协议
阅读更多...
【杂记-浅谈DHCP动态主机配置协议】

【杂记-浅谈DHCP动态主机配置协议】

DHCP动态主机配置协议 一、DHCP概述1、定义2、作用3、报文类型 二、DHCP的工作原理三、DHCP服务器的配置和管理 一、DHCP概述 1、定义 DHCP，Dynamic Host Configuration Protocol，动态主机配置协议，是一种网络协议，主要用于在IP网络中自动分配和管理IP地址以及其他网络配置参数。 2、作用 DHCP允许计算机和其他设备通
阅读更多...
常用MQ消息中间件Kafka、ZeroMQ和RabbitMQ对比及RabbitMQ详解

常用MQ消息中间件Kafka、ZeroMQ和RabbitMQ对比及RabbitMQ详解

1、概述   在现代的分布式系统和实时数据处理领域，消息中间件扮演着关键的角色，用于解决应用程序之间的通信和数据传递的挑战。在众多的消息中间件解决方案中，Kafka、ZeroMQ和RabbitMQ 是备受关注和广泛应用的代表性系统。它们各自具有独特的特点和优势，适用于不同的应用场景和需求。   Kafka 是一个高性能、可扩展的分布式消息队列系统，被设计用于处理大规模的数据流和实时数据传输。它
阅读更多...
JavaWeb系列六: 动态WEB开发核心(Servlet) 上

JavaWeb系列六: 动态WEB开发核心(Servlet) 上

韩老师学生 官网文档为什么会出现Servlet什么是ServletServlet在JavaWeb项目位置Servlet基本使用Servlet开发方式说明快速入门- 手动开发 servlet浏览器请求Servlet UML分析Servlet生命周期GET和POST请求分发处理通过继承HttpServlet开发ServletIDEA配置ServletServlet注意事项和细节 Servlet注
阅读更多...
【Unity Shader】片段着色器（Fragment Shader）的概念及其使用方法

【Unity Shader】片段着色器（Fragment Shader）的概念及其使用方法

在Unity和图形编程中，片段着色器（Fragment Shader）是渲染管线中的一个阶段，负责计算屏幕上每个像素（片段）的颜色和特性。片段着色器通常在顶点着色器和任何几何处理之后运行，是决定最终像素颜色的关键步骤。 Fragment Shader的概念： 像素处理：片段着色器处理经过顶点着色器和几何着色器处理后，映射到屏幕空间的像素。颜色计算：它计算每个像素的颜色值，这可能包括纹理采样、光
阅读更多...
【Unity Shader】Alpha Blend（Alpha混合）的概念及其使用示例

【Unity Shader】Alpha Blend（Alpha混合）的概念及其使用示例

在Unity和图形编程中，Alpha Blend（也称为Alpha混合）是一种用于处理像素透明度的技术。它允许像素与背景像素融合，从而实现透明或半透明的效果。Alpha Blend在渲染具有透明度的物体（如窗户、玻璃、水、雾等）时非常重要。 Alpha Blend的概念： Alpha值：Alpha值是一个介于0（完全透明）和1（完全不透明）的数值，用于表示像素的透明度。混合模式：Alpha B
阅读更多...
IPD推行成功的核心要素（十一）技术规划与平台规划促进公司战略成功

IPD推行成功的核心要素（十一）技术规划与平台规划促进公司战略成功

随着外部大环境的影响，各企业仅有良好的愿望是不够的。预测并顺应新兴市场和技术的变化，变危机为转机，不断推出强大的产品才是一个公司持续繁荣的根本保障。而高效的产品开发往往是基于某些关键技术，针对市场推出的一个或几个产品系列，这些产品系列通常共用一些产品平台，共用一种或者几种关键技术。当一家企业进入了平稳发展期，已经建立了较为完善的管理制度和产品开发流程，但是依然认为竞争对手是那样强大，那样不可战胜。
阅读更多...
2024年6月24日-6月30日（ue独立游戏为核心）

2024年6月24日-6月30日（ue独立游戏为核心）

试过重点放在独立游戏上，有个indienova独立游戏团队是全职的，由于他们干了几个月，节奏暂时跟不上，紧张焦虑了。五一时也有点自暴自弃了，实在没必要，按照自己的节奏走即可。精力和时间也有限，放在周末进行即可。除非哪天失业了，再也找不到工作了，再把重心放在独立游戏上。 另外，找到一个同样业余的美术，从头做肉鸽游戏，两周一次正式交流即可。节奏一定要放慢，不能影响正常工作生活。如果影响到了，还不如自
阅读更多...
RabbitMQ实践——临时队列

RabbitMQ实践——临时队列

临时队列是一种自动删除队列。当这个队列被创建后，如果没有消费者监听，则会一直存在，还可以不断向其发布消息。但是一旦的消费者开始监听，然后断开监听后，它就会被自动删除。 新建自动删除队列 我们创建一个名字叫queue.auto.delete的临时队列 绑定 我们直接使用默认交换器，所以不用创建新的交换器，也不用建立绑定关系。 实验 发布消息 我们在后台管理页面的默认交换器下向这个队列
阅读更多...
Spring MVC的核心类和注解——@RequestMapping注解（二）@RequestMapping注解的属性

Spring MVC的核心类和注解——@RequestMapping注解（二）@RequestMapping注解的属性

一、@RequestMapping注解的属性 属性名 类型 描述 name String 可选属性，用于为映射地址指定别名。 value String[] 可选属性，也是默认属性，用于指定请求的URL。 method RequestMethod[] 可选属性，用于指定该方法可以处理哪种类型的请求方式。 params String[] 可选属性，用于指定客户端请求中参数的值，必须包含
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2