蘇阿細 - RabbitMQ
type: channel
link: https://www.wangchouchou.com/category/RabbitMQ/
description: RabbitMQ
items: RabbitMQ - 其他补充
items: RabbitMQ - 发布确认补充
items: RabbitMQ - 延时队列
items: RabbitMQ - 死信队列
items: RabbitMQ - 交换机
items: RabbitMQ - 发布确认
items: RabbitMQ - WorkQueue
items: RabbitMQ - 介绍、四大核心概念
items: 分布式事务
items: RabbitMQ集群

RabbitMQ - 其他补充
type: item
date: 2023-06-23T13:24:00+00:00
description: 1. 幂等性（1）概念用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用。（2）消息重复消费消费者在消费 MQ 中的消息时，MQ 已把消息发送给消费者，消费者在给 MQ 返回 ack 时网络中断，故 MQ 未收到确认信息，该条消息会重新发给其他的消费者，或者在网络重连后再次发送给该消费者，但实际上该消费者已成功消费了该条消息，造成消费者消费了重复的消息。（3）解决思路一般使用全局 ID、唯一标识（时间戳、UUID）或 MQ 消息的 id tag 来判断，亦或是zi定义全局唯一 id，每次消费消息时先判断 id 对应的消息是否已消费过（4）消费端的幂等性保障在海量订单生成的业务高峰期，生产端有可能就会重复发送消息，这时候消费端就要实现幂等性，即消息永远不会被消费多次（即使多次收到了一样的消息）。唯一ID + 指纹码机制，利用数据库主键去重利用 redis 的原子性实现（setnx）2. 优先级队列（1）设置方式在管理页面添加在代码中添加Map<String, Object> arguments = new HashMap<>(); //优先级范围 0-10（设置过大会存在性能问题） arguments.put("x-max-priority", 10); channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);设置优先级之后的队列（2）测试Producerpublic class Producer { private static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.123.88"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("123456"); //创建连接 Connection connection = factory.newConnection(); //获取信道 Channel channel = connection.createChannel(); /** * 创建队列 * 参数说明： * 1.队列名称 * 2.是否持久化存储消息 * 3.是否进行消息共享（只供一个消费者进行消费） * 4.是否自动删除（最后一个消费者断开连接后，是否自动删除该队列） * 5.其他参数 */ /** * 2023-06-15偷懒 优先级队列 */ Map<String, Object> arguments = new HashMap<>(); //优先级范围 0-10（设置过大会存在性能问题） arguments.put("x-max-priority", 10); channel.queueDeclare(QUEUE_NAME, true, false, false, arguments); for (int i = 1; i < 10; i++) { String msg = "hello world" + i; if (i == 5) { AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build(); channel.basicPublish("", QUEUE_NAME, properties, msg.getBytes()); } else { channel.basicPublish("", QUEUE_NAME, null, msg.getBytes()); } } } }Consumerpublic class Consumer { private static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.123.88"); factory.setPort(5672); factory.setUsername("adm…

RabbitMQ - 发布确认补充
type: item
date: 2023-06-23T13:20:00+00:00
description: 1. 发布确认在RabbitMQ不可用的情况下，如何处理无法投递的消息？在application.yml配置文件中添加# 消息确认类型 执行回调 publisher-confirm-type: correlated补充：NONE 禁用发布确认模式，是默认值CORRELATED 发布消息成功到交换器后会触发回调方法SIMPLE 包含两种效果：和 CORRELATED 值一样会触发回调方法在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker（1）队列配置@Configuration public class ConfirmQueueConfig { public static final String CONFIRM_EXCHANGE = "confirm.exchange"; public static final String CONFIRM_QUEUE = "confirm.queue"; public static final String CONFIRM_ROUTING_KEY = "key1"; @Bean("confirmExchange") public DirectExchange confirmExchange() { return new DirectExchange(CONFIRM_EXCHANGE); } @Bean("confirmQueue") public Queue confirmQueue() { return QueueBuilder.durable(CONFIRM_QUEUE).build(); } @Bean public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue queue, @Qualifier("confirmExchange") Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(CONFIRM_ROUTING_KEY).noargs(); } }（2）消息生产者@Slf4j @RestController @RequestMapping("/confirm") public class SendConfirmMsgController { @Autowired private RabbitTemplate rabbitTemplate; @GetMapping("/sendMsg/{msg}/{routingKey}") @ApiOperation("发送消息（发布确认）") public void sendMsg(@PathVariable("msg") String msg, @PathVariable("routingKey") String routingKey) { log.info("当前时间：{}，发送消息：{}，routingKey：{}", new Date(), msg, routingKey); //正常发送 rabbitTemplate.convertAndSend(ConfirmQueueConfig.CONFIRM_EXCHANGE, routingKey, msg, new CorrelationData(UUID.randomUUID().toString())); } }（3）回调接口@Slf4j @Component public class ConfirmCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback { @Autowired private RabbitTemplate rabbitTemplate; /** * 注入 */ @PostConstruct public void init() { rabbitTemplate.setConfirmCallback(this); rabbitTemplate.setReturnCallback(this); } /** * 交换机确认回调方法 * * @param correlationData correlation data for the callback. 回调消息的ID及相关的信息 * @param …

RabbitMQ - 延时队列
type: item
date: 2023-06-22T15:26:00+00:00
description: 1. 概念队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素希望在指定时间到了之后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。2. 使用场景订单在十分钟之内未支付则自动取消新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒用户注册成功后，如果三天内没有登录则进行短信提醒用户发起退款，如果三天内没有得到处理则通知相关运营人员预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议……4. RabbitMQ中的TTLTTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这 条消息如果在 TTL 设置的时间内没有被消费，则会成为"死信"。如果同时配置了队列的 TTL 和消息的 TTL，则优先取较小的值。5. 设置TTL的两种方式（1）消息设置ttlrabbitTemplate.convertAndSend("X", "XC", msg, message -> { //设置ttl message.getMessageProperties().setExpiration(ttlTime); return message; });（2）队列设置ttl//ttl arguments.put("x-message-ttl", 1000 * 10); return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();（3）区别消息设置ttl，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间；另外还需要注意的一点是，如果不设置 ttl，表示消息永远不会过期，如果将 ttl设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃队列设置 ttl 时，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队 列中)6. 整合 SpringBootpom.xml<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.5.RELEASE</version> <relativePath/> </parent> <groupId>com.sw.rabbitmq</groupId> <artifactId>RabbitmqDemo01</artifactId> <version>0.0.1-SNAPSHOT</version> <name>RabbitmqDemo01</name> <description>RabbitmqDemo01</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>2.0.15</version> </dependency> <…

RabbitMQ - 死信队列
type: item
date: 2023-06-22T15:23:00+00:00
description: 无法被消费的消息，即：producer 将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中1. 来源消息 TTL 过期队列达到最大长度(队列满了，无法再添加数据到 mq 中)消息被拒绝(basic.reject 或 basic.nack)并且 requeue = false2. 消息 TTL 过期Producerpublic class Producer { //普通交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); //死信消息 AMQP.BasicProperties properties = new AMQP.BasicProperties() .builder().expiration("10000").build(); for (int i = 1; i < 11; i++) { String msg = "info" + i; channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, msg.getBytes()); System.out.println("发送的消息是：" + msg); } } }Consumer1 （启动之后关闭，模拟消费者不能正常消息的情况）public class Consumer01 { //普通交换机 public static final String NORMAL_EXCHANGE = "normal_exchange"; //死信交换机 private static final String DEAD_EXCHANGE = "dead_exchange"; //普通队列 public static final String NORMAL_QUEUE = "normal_queue"; //死信队列 private static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); //声明普通/死信交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); //声明普通队列 //普通队列绑定死信交换机 Map<String, Object> arguments = new HashMap<>(); //过期时间（过期时间可由生产者或者消费者设置，二选一） arguments.put("x-message-ttl", 10000); //死信交换机 arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE); //死信routingKey arguments.put("x-dead-letter-routing-key", "lisi"); channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments); //死信队列 channel.queueDeclare(DEAD_QUEUE, false, false, false, null); //绑定普通交换机 - 队列 channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan"); //绑定死信交换机 - 队列 channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi"); System.out.println("消费者c1…

RabbitMQ - 交换机
type: item
date: 2023-06-20T14:13:00+00:00
description: 1. Exchanges（1）概念RabbitMQ 消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。实际上，生产者甚至都不知道这些消息传递传递到了哪些队列中。 相反，生产者只能将消息发送到交换机（exchange），交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息，是应该把这些消息放到特定队列还是说把他们放到许多队列中，亦或是应该丢弃它们，这就得由交换机的类型来决定。（2）交换机类型直接（direct）主题（topic）标题（headers）扇出（fanout）（3）无名exchangechannel.basicPublish("", QUEUE_NAME, properties, msg.getBytes());之前的demo中生产者发送消息时未设置交换机，即第一个参数为空，表示使用的是默认交换机（又称无名交换机）2. 临时队列在之前的demo中使用的都是临时队列，一旦断开了消费者的连接，队列将被自动删除创建临时队列的方式：String queue = channel.queueDeclare().getQueue();3. 绑定binding 是 exchange 和 queue 之间的桥梁，即 exchange 和哪个队列进行了绑定4. Fanout（1）介绍将接收到的所有消息广播到它知道的所有队列中，RabbitMQ默认的交换机如下：（2）DemoProducerpublic class EmitLog { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()) { String msg = scanner.next(); channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes(StandardCharsets.UTF_8)); System.out.println("发送的消息是：" + msg); } } }Consumerpublic class ReceiveLog01 { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //生成临时队列 String queueName = channel.queueDeclare().getQueue(); //绑定临时队列，其中routingKey为空字符串 channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println("ReceiveLog01等待接收消息，接收到消息后在控制台打印..."); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String msg = new String(delivery.getBody(), StandardCharsets.UTF_8); System.out.println("接收到的消息是：" + msg); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } } public class ReceiveLog02 { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = …

RabbitMQ - 发布确认
type: item
date: 2023-06-20T14:12:00+00:00
description: 1. 发布确认原理生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的 消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置 basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。注：confirm 模式是异步的2. 发布确认策略（1）开启发布确认发布确认默认未开启，手动开启//开启发布确认 channel.confirmSelect();（2）单个确认发布是一种同步确认发布的方式，即发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirmsOrDie(long) 这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。这种确认方式有一个最大的缺点就是：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量，但对于某些应用程序来说这可能已经足够了。public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); //开始时间 long startTime = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String msg = i + ""; channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes()); //单个确认 boolean flag = channel.waitForConfirms(); if (flag) { System.out.println("第" + i + "条消息发送成功"); } } //开始时间 long endTime = System.currentTimeMillis(); System.out.println("发布" + MESSAGE_COUNT + "条数据（单个确认模式）耗时：" + (endTime - startTime)); }（3）批量发布确认先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是：当发生故障导致发布出现问题时，不知道是哪个消息出现了问题，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息，该方案也是同步的，也一样阻塞消息的发布。public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException { Channel channel = RabbitMqUtils.getChannel(); String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); int batchSize = 100; //开始时间 long startTime = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String msg = i + ""; channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes())…

RabbitMQ - WorkQueue
type: item
date: 2023-06-19T14:30:00+00:00
description: 工作队列（又称任务队列）：其主要思想是避免立即执行资源密集型任务，而不得不等待它完成。消息队列将任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。1. 轮询分发（1）抽取工具类public class RabbitMqUtils { /** * 获取Channel * * @return Channel * @throws IOException * @throws TimeoutException */ public static Channel getChannel() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.123.88"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("123456"); Connection connection = factory.newConnection(); return connection.createChannel(); } }（2）工作线程public class Worker01 { private static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("收到消息：" + new String(message.getBody())); CancelCallback cancelCallback = (consumerTag) -> System.out.println(consumerTag + "消费者取消消费接口回调逻辑"); //接收消息 System.out.println("c1等待接收消息..."); channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback); } } public class Worker02 { private static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("收到消息：" + new String(message.getBody())); CancelCallback cancelCallback = (consumerTag) -> System.out.println(consumerTag + "消费者取消消费接口回调逻辑"); //接收消息 System.out.println("c2等待接收消息..."); channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback); } }（3）Task线程public class Task01 { private static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMqUtils.getChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()) { Stri…

RabbitMQ - 介绍、四大核心概念
type: item
date: 2023-06-19T14:29:00+00:00
description: 参考于b站尚硅谷RabbitMQ课程一、介绍RabbitMQ 是一个消息中间件：它接受并转发消息，可以把它当做一个快递站点，当你要发送一个包裹时，你把包裹放到快递站，快递员最终会把你的快递送到收件人那里，此时 RabbitMQ 可以比作是一个快递站，由快递员帮你传递快件（RabbitMQ 与快递站的主要区别在于它不处理快件而仅作接收，存储和转发消息数据）二、四大核心概念生产者产生数据发送消息交换机交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息 推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推 送到多个队列，亦者是将消息丢弃，这个得由交换机类型决定 队列队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中，队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。多个生产者可以将消息发送到一个队列，同理多个消费者也可以尝试从一个队列接收数据消费者消费与接收具有相似的含义，消费者大多时候是一个等待接收消息的程序。注：生产者、消费者和消息中间件很多时候并不在同一台机器上，同一个应用程序既可以是生产者又可以是消费者名词介绍Broker：接收和分发消息的应用，RabbitMQ Server 就是 Message BrokerVirtual host：出于多租户和安全因素设计，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出 多个 vhost，每个用户在自己的 vhost 创建 exchange／queue 等 Connection：publisher／consumer 和 broker 之间的 TCP 连接Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection 的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个 thread 创建单独的 channel 进行通讯，AMQP method 包含了 channel id 帮助客户端和 message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到 queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe), fanout (multicast)Queue：消息最终被送到这里等待 consumer 取走Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key，Binding 信息被保 存到 exchange 中的查询表中，用于 message 的分发依据安装可参考官网文档：https://www.rabbitmq.com/download.htmlDemopom.xml<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.sw</groupId> <artifactId>rabbitmq</artifactId> <version>1.0-SNAPSHOT</version> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properti…

分布式事务
type: item
date: 2021-04-18T06:18:00+00:00
description: 一、概述指事务的操作位于不同的节点上，需要保证事务的ACID原则，如：订单与库存不在同一个节点上，但需要保证两个事务的一致性（下单—减少库存，取消订单—库存恢复）二、实现分布式事务的方式1、两段提交（2PC）需要数据库厂商的支持，java组件有atomikos等两阶段提交（Two-phase Commit），通过引入协调者（Coordinator）来协调参与者的行为，并最终决定这些参与者是否要真正执行事务准备阶段协调者询问参与者事务是否执行成功，参与者返执行回结果提交阶段如果事务在每个参与者上都执行成功，事务协调者发送通知让参与者提交事务；否则反之，即回滚事务注：参与者在准备阶段执行成功，但未提交事务，只有在提交阶段收到协调者的消息，才会执行提交或回滚存在的问题1、同步阻塞，所有参与者在等待其他参与者相应的时候都处于同步阻塞状态，无法进行其他操作2、单点问题，协调者在2PC中起到非常大的作用，产生故障时影响也非常大3、数据不一致，在阶段二，如果协调者只发送了部分Commit消息，此时如果产生网络异常，则只有接收到Commit消息的参与者提交了事务，使得数据不一致4、任意一个节点失败，就会导致整个事务失败，容错机制不完善2、补偿事务（TCC）针对每个操作，都注册一个与其对应的确认和补偿（撤销）操作，分为三个阶段：Try 对业务系统做检测及资源预留Confirm 对业务系统做确认提交，Try阶段执行成功并开始执行Confirm阶段时，默认Confirm阶段是不会出错的（只要Try成功，Confirm一定成功）Cancel 在业务执行错误，需要回滚时执行业务的取消，释放预留资源优缺点实现流程相较于2PC简单，但数据一致性比2PC要更差一些TCC属于应用层的一种补偿方式，所以在实现业务的时候需要编写很多的补偿代码，在一些业务场景中，业务处理不太友好3、本地消息表（异步确保）本地消息表与业务数据表处于同一个数据库中，这样就能利用本地事务来保证在对这两个表的操作满足事务特性，并且使用消息队列保证最终一致性。在分布式事务操作的一方完成写业务数据的操作之后向本地消息表发送一个消息，本地事务能保证这个消息一定会被写入本地消息表中之后将本地消息表的消息转发到Kafka等消息队列中，如果转发成功则将消息从本地消息表中删除，否则继续重新转发在分布式事务操作的另一方从消息队列中读取一个消息，并执行消息中的操作优缺点1、避免了分布式事务，实现数据一致性2、消息表会耦合到业务系统中，耦合度太高4、MQ事务消息一些第三方的MQ是支持事务消息的，如RocketMQ，它们支持事务的方式类似于二阶段提交，流程大致为：第一阶段：Prepared消息，拿到消息的地址；第二阶段：执行本地事务；第三阶段：通过第一阶段拿到的地址去访问消息，并修改状态也就是说在业务方法内要让消息队列提交两次请求，一次发送消息，一次确认消息。如果确认消息发送失败了，RabbitMQ会定期扫描消息集群中的事务消息，这时候发现了Prepared消息，它会向消息发送者确认，所以生产方需要实现一个check接口，RabbitMQ会根据发送端设置的策略来决定回滚还是继续发送确认消息，这样就保证了消息发送与本地事务同时成功或同时失败优缺点1、实现最终一致性，不需要依赖本地数据库事务2、实现难度大，主流MQ不支持

RabbitMQ集群
type: item
date: 2021-04-18T06:17:00+00:00
description: 以单机多节点搭建为例1、启动第一个节点 rabbit-1RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server start &2、启动第二个节点 rabbit-2注：web管理插件的端口号需要另外指定，否则会提示默认的 5672 端口占用RABBITMQ_NODE_PORT=5673 RABBIT_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server start &3、查看服务ps aux|grep rabbitmq4、设置rabbit-1为主节点# 停止应用 rabbitmqctl -n rabbit-1 stop_app # 重置，清除节点上的历史信息 rabbitmqctl -n rabbit-1 reset # 启动 rabbitmqctl -n rabbit-1 start_app5、设置从节点# 停止应用 rabbitmqctl -n rabbit-2 stop_app # 重置，清除节点上的历史信息 rabbitmqctl -n rabbit-2 reset # 将rabbit-2加入到rabbit-1主节点 # 'Server-node' 主节点名，实际操作时需更换为主机名 rabbitmqctl -n rabbit-2 join_cluster rabbit-1@'Server-node' # 启动 rabbitmqctl -n rabbit-2 start_app6、验证集群状态rabbitmqctl cluster_status -n rabbit-17、web管理给node节点设置用户名和密码rabbitmqctl -n rabbit-1 add_user admin 123456 rabbitmqctl -n rabbit-1 set_user_tags admin administrator rabbitmqctl -n rabbit-1 set_permissions -p / admin ".*" ".*" ".*" # 其他节点以此类推补充如果采用多机部署集群的方式，需读取其中一个节点的cookie，并复制到其他节点（节点之间通过cookie确认相互间是否能进行通信）cookie存放路径：/var/lib/rabbitmq/.erlang.cookie