消息队列-RabbitMQ（二）

接上文《消息队列-RabbitMQ（一）》

@Configuration
public class RabbitMqConfig {
    // 消息的消费方json数据的反序列化
    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        return factory;
    }

    // 定义使用json的方式转换数据
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate amqpTemplate = new RabbitTemplate();
        amqpTemplate.setConnectionFactory(connectionFactory);
        amqpTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return amqpTemplate;
    }

}

@Service
public class TestConsumer {

    @RabbitListener(queuesToDeclare = {@Queue("simpleQueue")})
    public void simpleModel(User user) {
        System.out.println("接收消息message=" + user);
    }

    // value=@Queue 创建临时队列
    // exchange创建交换机
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(value = "fanout-ex", type = ExchangeTypes.FANOUT))
    })
    public void receiveMessage1(User user) {
        System.out.println(String.format("消费者 【one】: %s", user));
    }

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue,
                    exchange = @Exchange(value = "fanout-ex", type = ExchangeTypes.FANOUT))
    })
    public void receiveMessage2(User user) {
        System.out.println(String.format("消费者 【two】: %s", user));
    }

}

@RestController
@RequestMapping("/rabbitMqReq")
public class RabbitMqController {
    @Autowired
    private TestProducer testProducer;

    @PostMapping("/simple")
    public void simple(@RequestBody User user) {
        testProducer.simpleMessageSend();
    }

    @PostMapping("/fanoutMessageSend")
    public void fanoutMessageSend(@RequestBody User user) {
        testProducer.fanoutMessageSend();
    }
}

@Data
public class User {
    private String userId; //用户编号
    private String userName; //用户姓名
}

@Service
public class TestProducer {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void simpleMessageSend() {
        System.out.println("simpleMessageSend");
        User user = new User();
        user.setUserId("1001");
        user.setUserName("张三");
        rabbitTemplate.convertAndSend("simpleQueue", user);
    }

    // fanout模型
    public void fanoutMessageSend() {
        for (int i = 0; i < 5; i++) {
            User user = new User();
            user.setUserId("1001");
            user.setUserName("张三");
            rabbitTemplate.convertAndSend("fanout-ex", "", user);
        }
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })
public class App
{
    public static void main(String[] args)
    {
        SpringApplication.run(App.class, args);
        System.out.println("(♥◠‿◠)ﾉﾞ  启动成功   ლ(´ڡ`ლ)ﾞ ");
    }
}

# 开发环境配置
server:
  # 服务器的HTTP端口，默认为8080
  port: 8899
  servlet:
    # 应用的访问路径
    context-path: /
  tomcat:
    # tomcat的URI编码
    uri-encoding: UTF-8
    # tomcat最大线程数，默认为200
    max-threads: 800
    # Tomcat启动初始化的线程数，默认值25
    min-spare-threads: 30

<?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>org.example</groupId>
  <artifactId>RabbitMQDemo</artifactId>
  <version>1.0-SNAPSHOT</version>

  <name>RabbitMQDemo</name>
  <!-- FIXME change it to the project's website -->
  <url>http://www.example.com</url>

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.1.RELEASE</version>
    <relativePath />
  </parent>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
    <fastjson.version>1.2.70</fastjson.version>
  </properties>

  <dependencies>
    <!-- SpringBoot 核心包 -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>

    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- AMQP客户端 -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

    <!-- 阿里JSON解析器 -->
    <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>${fastjson.version}</version>
    </dependency>

    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
      <version>1.18.10</version>
    </dependency>

  </dependencies>

  <repositories>
    <!--阿里云镜像仓库-->
    <repository>
      <id>public</id>
      <name>aliyun nexus</name>
      <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
      <releases>
        <enabled>true</enabled>
      </releases>
    </repository>
    <!--oracle驱动没有发布到中央仓库，只能从此仓库下载-->
    <repository>
      <id>jeecg</id>
      <name>jeecg Repository</name>
      <url>http://maven.jeewx.com/nexus/content/repositories/jeecg</url>
      <snapshots>
        <enabled>false</enabled>
      </snapshots>
    </repository>
  </repositories>

</project>

什么是事务？
数据库事务。原子性、一致性、隔离性、持久性。

本地事务？
整个服务操作只能涉及一个单一的数据库资源。

分布式事务？
分布式事务指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。分布式事务就是为了保证不同数据库的数据一致性。

分布式事务应用架构？
微服务架构的分布式应用环境下，越来越多的应用要求对多个数据库资源，多个服务的访问都能纳入到同一个事务当中，分布式事务应运而生。

分布式事务有几种解决方案： 两阶段提交/XA MQ
第一阶段（prepare）：即所有的参与者RM准备执行事务并锁住需要的资源。参与者ready时，向TM报告已准备就绪。
第二阶段 (commit/rollback)：当事务管理者™确认所有参与者(RM)都ready后，向所有参与者发送commit命令。

https://blog.csdn.net/qq_43410878/article/details/123656765 RabbitMQ使用详解

RabbitMQ与springboot整合
1、添加依赖 spring-boot-starter-amqp
2、配置RabbitConfig，包含RabbitListenerContainerFactory、RabbitTemplate

简单模型
@SpringBootTest
class SpringbootRabbitmqApplicationTests {
@Resource
private RabbitTemplate rabbitTemplate;

@Test
public void simpleMessageSend() {
    rabbitTemplate.convertAndSend("simpleQueue", new User(1, "张"));
}

}

@RabbitListener(queuesToDeclare = {@Queue(“simpleQueue”)})
public void simpleModel(User user) {
log.info(“message: {}”, user);
}

发布订阅模型
// fanout模型
@Test
public void fanoutMessageSend() {
for (int i = 0; i < 5; i++) {
rabbitTemplate.convertAndSend(“fanout-ex”, “”, new User(i, “张三”));
}
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

// value=@Queue 创建临时队列
// exchange创建交换机
@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
exchange = @Exchange(value = “fanout-ex”, type = ExchangeTypes.FANOUT))
})
public void receiveMessage1(User user) {
System.out.println(String.format(“消费者 【one】: %s”, user));
}

@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
exchange = @Exchange(value = “fanout-ex”, type = ExchangeTypes.FANOUT))
})
public void receiveMessage2(User user) {
System.out.println(String.format(“消费者 【two】: %s”, user));
}

直连模式(direct)
@Test
public void directMessageSend() {
//rabbitTemplate.convertAndSend(“direct-ex”, “success”, new User(2, “张三”));
rabbitTemplate.convertAndSend(“direct-ex”, “error”, new User(2, “张三”));
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
key = {“error”, “success”},
exchange = @Exchange(value = “direct-ex”, type = ExchangeTypes.DIRECT))
})
public void receiveMessage1(User user) {
System.out.println(String.format(“消费者 【one】: %s”, user));
}

@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
key = {“error”},
exchange = @Exchange(value = “direct-ex”, type = ExchangeTypes.DIRECT))
})
public void receiveMessage2(User user) {
System.out.println(String.format(“消费者 【two】: %s”, user));
}

topic模型
@Test
public void topicMessageSend() {
//rabbitTemplate.convertAndSend(“topic-ex”, “company”, new User(2, “张三”));
rabbitTemplate.convertAndSend(“topic-ex”, “company.java”, new User(2, “张三”));
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
key = {“company.#”},
exchange = @Exchange(value = “topic-ex”, type = ExchangeTypes.TOPIC))
})
public void receiveMessage1(User user) {
System.out.println(String.format(“消费者 【one】: %s”, user));
}

@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
key = {“company.java.#”},
exchange = @Exchange(value = “topic-ex”, type = ExchangeTypes.TOPIC))
})
public void receiveMessage2(User user) {
System.out.println(String.format(“消费者 【two】: %s”, user));
}

@RabbitListener(bindings = {
@QueueBinding(value = @Queue,
key = {“company.html.*”},
exchange = @Exchange(value = “topic-ex”, type = ExchangeTypes.TOPIC))
})
public void receiveMessage3(User user) {
System.out.println(String.format(“消费者 【three】: %s”, user));
}

消息的手动确认
spring:
rabbitmq:
listener:
simple:
# 提交方式为手动
acknowledge-mode: MANUAL

factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

面试：a. 如何保证消息不丢失？

​ b. 如何保证消息的不重复消费？

​ c. 如何使用mq来是实现分布式事务？

​ d. 在工作中mq用在哪里？支付回调。

RabbitMQ如何做到消息不丢失？

1.持久化

发送消息时设置delivery_mode属性为2，使消息被持久化保存到磁盘，即使RabbitMQ服务器宕机也能保证消息不丢失。同时，创建队列时设置durable属性为True，以确保队列也被持久化保存。

2.确认机制

消费者通过basic.ack命令向RabbitMQ服务器确认已经消费了消息。如果消费者处理消息时发生错误或宕机，RabbitMQ会重新将消息发送给其他消费者。RabbitMQ在接收到消费者确认消息前会将消息保存在内存中，在确认后才会删除消息。

3.发布者确认

RabbitMQ支持发布者确认机制，即发布者在将消息发送到队列后，等待RabbitMQ服务器的确认消息。成功保存到队列的消息会返回确认消息给发布者，如果无法保存则返回Nack（Negative Acknowledgement）消息。通过发布者确认机制，可以确保消息成功发送到RabbitMQ服务器。

4.备份队列

RabbitMQ支持备份队列（Alternate Exchange）机制，即在消息发送到队列之前，先将消息发送到备份队列。如果主队列无法接收消息，RabbitMQ会将消息发送到备份队列中。备份队列通常是一个交换机，在创建队列时可以通过x-dead-letter-exchange属性指定备份队列。

三、注意事项 在使用RabbitMQ消息确认机制时，需注意以下几点：

1、确认模式的选择： 根据具体业务需求选择合适的确认模式。如果对消息传递的可靠性要求较高，建议使用手动确认模式或批量确认模式。

2、设置消息持久化： 在发布消息时，通过设置消息的持久化属性，可以确保即使RabbitMQ服务器意外关闭或重启，消息仍然能够被保存下来。

3、处理未确认消息： 如果消费者在处理消息时发生异常，导致无法发送确认信号给RabbitMQ，那么这些消息将保持在队列中，并可能被重新投递。需要设定适当的重试策略和错误处理机制来处理这些未确认的消息。

4、监听确认超时： 在使用批量确认模式时，如果长时间没有收到确认信号，需要设置合理的超时时间，并采取相应措施来处理超时的情况。

RabbitMQ消息确认机制详解：确保交付成功

消息确认机制是通过发布者（Producer）和消费者（Consumer）之间的交互来实现的。
当发布者发送消息到RabbitMQ后，会等待确认结果。如果消息成功被消费者接收并处理，消费者会发送一个确认信号给RabbitMQ，告知消息已经处理完成。而RabbitMQ则会根据接收到的确认信号，判断消息是否成功交付。
消息确认机制一般有两种模式：简单模式和批量模式。在简单模式中，每发送一条消息后就等待确认；而在批量模式中，可以一次发送多条消息，然后等待批量确认。无论是简单模式还是批量模式，都可以提高消息传递的可靠性。

https://blog.csdn.net/weixin_52278591/article/details/128841155 RabbitMq 消息确认机制详解

https://blog.csdn.net/u011942456/article/details/128198956 总结RabbitMq消息丢失和消息重复消费问题

https://www.zhihu.com/question/54756562?utm_id=0 rabbitmq是什么，主要用于哪些方面？

https://zhuanlan.zhihu.com/p/583520436?utm_id=0 用了8年MQ！聊聊消息队列的技术选型，哪个最香！

https://blog.csdn.net/qq_43410878/article/details/123656765 RabbitMQ使用详解

https://zhuanlan.zhihu.com/p/590948541 绝对详细的 RabbitMQ入门，看完本系列就够了

MQ的优势和劣势

https://cloud.tencent.com/developer/article/2113514 面试百问：使用MQ的优势、劣势以及问题

rabbitmq basicAck