<?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>
    <!--注意：spring-boot-starter-parent版本若达到3.0.0 则只支持JDK17，而2.7.10是支持JDK8的最后一个版本-->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.10</version>
    </parent>
 
    <groupId>cn.xw</groupId>
    <artifactId>HelloWorld</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>HelloWorld</name>
    <description>HelloWorld</description>
 
    <dependencies>
        <!--Spring Boot的核心启动器，包含了自动配置、日志和YAML-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
            <!--去掉logback配置，要不然冲突-->
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!-- 引入SpringBoot的log4j2依赖启动坐标；这坐标包含具体的log4j2的坐标和连接Slf4j的适配器 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-log4j2</artifactId>
        </dependency>
        <!--SpringBootWeb启动依赖坐标-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--RabbitMQ启动依赖坐标-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!-- JSON格式化坐标-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.26</version>
        </dependency>
        <!--Lombok坐标导入-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.26</version>
            <scope>provided</scope>
        </dependency>
        <!-- RabbitMQ测试坐标 -->
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <version>3.0.3</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <!--配置maven编译版本-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source><!--源代码使用的JDK-->
                    <target>1.8</target><!--target需要生成的目标class文件的编译版本-->
                    <encoding>UTF-8</encoding><!--字符集编码，防止中文乱码-->
                    <failOnError>true</failOnError><!--指示即使存在编译错误，构建是否仍将继续-->
                    <failOnWarning>false</failOnWarning><!--指示即使存在编译警告，构建是否仍将继续-->
                    <showDeprecation>false</showDeprecation><!--设置是否显示使用不推荐API的源位置-->
                    <showWarnings>false</showWarnings><!--设为true若要显示编译警告，请执行以下操作-->
                    <meminitial>128M</meminitial><!--编译器使用的初始化内存-->
                    <maxmem>512M</maxmem><!--编译器使用的最大内存-->
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

<?xml version="1.0" encoding="UTF-8" ?>
<!--monitorInterval属性值（秒数）为一个非零值来让Log4j每隔指定的秒数来重新读取配置文件，可以用来动态应用Log4j配置-->
<Configuration status="info" monitorInterval="30">
    <!--用来自定义一些变量-->
    <Properties>
        <!--变量定义-->
        <Property name="myPattern" value="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
        <Property name="dir_url">./logs</Property>
    </Properties>
    <!--使用Appenders元素可以将日志事件数据写到各种目标位置-->
    <Appenders>
        <!-- 默认打印到控制台 -->
        <Console name="ConsoleAppend" target="SYSTEM_OUT">
            <!-- 默认打印格式 -->
            <PatternLayout pattern="${myPattern}"/>
        </Console>
        <!-- 打印到日志文件上 -->
        <File name="FileAppend" fileName="${dir_url}/fileLog.log" bufferedIO="true" immediateFlush="true">
            <PatternLayout>
                <pattern>${myPattern}</pattern>
            </PatternLayout>
        </File>
    </Appenders>
    <!--定义logger，只有定义了logger并引入的appender，appender才会生效-->
    <Loggers>
        <!-- 默认打印日志级别为 error -->
        <Root level="INFO">
            <AppenderRef ref="ConsoleAppend"/>
            <AppenderRef ref="FileAppend"/>
        </Root>
    </Loggers>
</Configuration>

server:
  port: 8081
spring:
  ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties
  rabbitmq:
    host: 49.235.99.193
    port: 5672
    username: admin
    password: 123
    virtual-host: test

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-14 11:49
 * 用来测试RabbitMQ的生产者发送消息（对象）到消费者中的一系列传输
 */
@Data
public class MessageSendDTO implements Serializable {
    private static final long serialVersionUID = 5905249092659173678L;
    private Integer msgID;          // 消息ID
    private String msgType;         // 消息类型
    private Object msgBody;         // 消息体
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-13 17:55
 * RabbitMQ配置类
 */
@Configuration
public class RabbitMQConfig {
    //定义1：简单的直接交换机名称；定义2：简单队列名称；定义3：路由key
    public static final String SIMPLE_DIRECT_EXCHANGE = "simpleDirectExchange";
    public static final String SIMPLE_QUEUE_NAME = "simpleQueueName";
    public static final String SIMPLE_KEY = "simpleKey";
    /***
     * 创建交换机信息
     * @return Exchange
     */
    @Bean("simpleDirectExchange")   //注：Bean对象可以不写名称，默认就是方法名
    public Exchange simpleDirectExchange() {
        //这个ExchangeBuilder就是我们当初使用的如下方式一样：
        // channel.exchangeDeclare("交换机名称", "交换机类型",true, false, false, null);
        return ExchangeBuilder.directExchange(SIMPLE_DIRECT_EXCHANGE).durable(true).build();
    }
    /***
     * 创建队列信息
     * @return Queue
     */
    @Bean("simpleQueueName")
    public Queue simpleQueueName() {
        //这个QueueBuilder就是我们当初使用的如下方式一样：
        // channel.queueDeclare("队列名称", true, false, false, null);
        return QueueBuilder.durable(SIMPLE_QUEUE_NAME).build();
    }
    /***
     * 队列信息绑定到交换机上
     * @param simpleDirectExchange 简单的直接交换机
     * @param simpleQueueName 简单的队列
     * @return Binding
     */
    @Bean("simpleQueueBindSimpleExchange")
    public Binding simpleQueueBindSimpleExchange(@Qualifier(value = "simpleDirectExchange") Exchange simpleDirectExchange,
                                                 @Qualifier(value = "simpleQueueName") Queue simpleQueueName) {
        //这个BindingBuilder就是我们当初使用的如下方式一样：
        // channel.queueBind("队列名称", "交换机名称", "路由key");
        return BindingBuilder.bind(simpleQueueName).to(simpleDirectExchange).with(SIMPLE_KEY).noargs();
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-13 23:52
 */
@Slf4j  //使用lombok自带的日志注解，具体实现是slf4j+log4j2
@RestController
@RequestMapping("/simple")
@RequiredArgsConstructor
public class SimpleController {
    //使用SLF4J来获取Logger对象;(注意导包：import org.slf4j.Logger; import org.slf4j.LoggerFactory;)
    //Logger logger = LoggerFactory.getLogger(this.getClass());
    //注入生产者对象
    private final SimpleProducer simpleProducer;
    /***
     * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息
     * @param msg 请求消息
     * @return String
     */
    @PostMapping("/produce")
    public String produce(@RequestBody MessageSendDTO msg) {
        log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg);
        //发送消息到生产者
        simpleProducer.productionSimpleMessage(msg);
        return "请求发送成功，并已接收";
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-13 23:30
 * 生产者
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SimpleProducer {
    //注入RabbitTemplate对象
    private final RabbitTemplate rabbitTemplate;
    /***
     * 生产者方法
     * @param msg 生产者发送的消息
     */
    public void productionSimpleMessage(MessageSendDTO msg) {
        log.info("生产者接收到消息，并发送到Brock的交换机....");
        //消息转换为JSON格式发送，并发送到Brock的交换机
        byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8);
        //convertAndSend("交换机名称","路由key","发送消息内容")，其实和原生的：
        // channel.basicPublish("交换机名称","路由key","其它参数","消息");
        // 使用convertAndSend默认消息是持久化的，如我们当初原生设置的 其它参数：MessageProperties.PERSISTENT_TEXT_PLAIN
        rabbitTemplate.convertAndSend(RabbitMQConfig.SIMPLE_DIRECT_EXCHANGE, RabbitMQConfig.SIMPLE_KEY, bytes);
    }
}
//----------------------------------------------------------------------
 
/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-13 23:29
 * 这是一个简单的消费者
 */
@Slf4j
@Component
public class SimpleConsumer {
    /***
     * 简单消息处理（监听）
     * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型
     * @param message 这个就类似我们原生的message
     * @param channel 这个就类似我们原生的channel
     */
    @RabbitListener(queues = {RabbitMQConfig.SIMPLE_QUEUE_NAME}) //只需要监听队列即可,多个则在{}里面逗号分割
    public void messageSimpleHandle(String msgData, Message message, Channel channel) {
        //获取到队列消息，因为发送是JSON格式，我们要解析对象格式
        MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class);
        log.info("消息由消费者消费：{}，并消费完成", msg);
    }
}

## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties，以实际版本为主
# base
spring.rabbitmq.host: 服务Host
spring.rabbitmq.port: 服务端口
spring.rabbitmq.username: 登陆用户名
spring.rabbitmq.password: 登陆密码
spring.rabbitmq.virtual-host: 连接到rabbitMQ的vhost
spring.rabbitmq.addresses: 指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host)
spring.rabbitmq.requested-heartbeat: 指定心跳超时，单位秒，0为不指定；默认60s
spring.rabbitmq.publisher-confirm-type: 是否启用【发布确认】
spring.rabbitmq.publisher-returns: 是否启用【发布返回】
spring.rabbitmq.connection-timeout: 连接超时，单位毫秒，0表示无穷大，不超时
spring.rabbitmq.parsed-addresses:
# ssl
spring.rabbitmq.ssl.enabled: 是否支持ssl
spring.rabbitmq.ssl.key-store: 指定持有SSL certificate的key store的路径
spring.rabbitmq.ssl.key-store-password: 指定访问key store的密码
spring.rabbitmq.ssl.trust-store: 指定持有SSL certificates的Trust store
spring.rabbitmq.ssl.trust-store-password: 指定访问trust store的密码
spring.rabbitmq.ssl.algorithm: ssl使用的算法，例如，TLSv1.1
# cache
spring.rabbitmq.cache.channel.size: 缓存中保持的channel数量
spring.rabbitmq.cache.channel.checkout-timeout: 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel
spring.rabbitmq.cache.connection.size: 缓存的连接数，只有是CONNECTION模式时生效
spring.rabbitmq.cache.connection.mode: 连接工厂缓存模式：CHANNEL 和 CONNECTION
# listener
spring.rabbitmq.listener.simple.auto-startup: 是否启动时自动启动容器
spring.rabbitmq.listener.simple.acknowledge-mode: 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto
spring.rabbitmq.listener.simple.concurrency: 最小的消费者数量
spring.rabbitmq.listener.simple.max-concurrency: 最大的消费者数量
spring.rabbitmq.listener.simple.prefetch: 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量.
spring.rabbitmq.listener.simple.transaction-size: 指定一个事务处理的消息数量，最好是小于等于prefetch的数量.
spring.rabbitmq.listener.simple.default-requeue-rejected: 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
spring.rabbitmq.listener.simple.idle-event-interval: 多少长时间发布空闲容器时间，单位毫秒
spring.rabbitmq.listener.simple.retry.enabled: 监听重试是否可用
spring.rabbitmq.listener.simple.retry.max-attempts: 最大重试次数
spring.rabbitmq.listener.simple.retry.initial-interval: 第一次和第二次尝试发布或传递消息之间的间隔
spring.rabbitmq.listener.simple.retry.multiplier: 应用于上一重试间隔的乘数
spring.rabbitmq.listener.simple.retry.max-interval: 最大重试时间间隔
spring.rabbitmq.listener.simple.retry.stateless: 重试是有状态or无状态
# template
spring.rabbitmq.template.mandatory: 启用强制信息；默认false
spring.rabbitmq.template.receive-timeout: receive() 操作的超时时间
spring.rabbitmq.template.reply-timeout: sendAndReceive() 操作的超时时间
spring.rabbitmq.template.retry.enabled: 发送重试是否可用
spring.rabbitmq.template.retry.max-attempts: 最大重试次数
spring.rabbitmq.template.retry.initial-interval: 第一次和第二次尝试发布或传递消息之间的间隔
spring.rabbitmq.template.retry.multiplier: 应用于上一重试间隔的乘数
spring.rabbitmq.template.retry.max-interval: 最大重试时间间隔

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-14 17:43
 * RabbitMQ配置类
 */
@Configuration
public class RabbitMQConfig {
    //直接交换机名称
    public static final String DIRECT_EXCHANGE = "directExchange";
    //队列名称A
    public static final String QUEUE_A_NAME = "queueAName";
    //路由key
    public static final String ROUTE_KEY = "routeKeyName";
    /***
     * 创建交换机信息
     * @return Exchange
     */
    @Bean("directExchange")
    public Exchange directExchange() {
        return ExchangeBuilder.directExchange(DIRECT_EXCHANGE).durable(true).build();
    }
    /***
     * 创建队列A信息
     * @return Queue
     */
    @Bean("queueAName")
    public Queue queueAName() {
        return QueueBuilder.durable(QUEUE_A_NAME).build();
    }
    /***
     * 队列绑定到交换机上，通过路由key
     * @param directExchange 交换机信息
     * @param queueAName A队列绑定
     * @return Binding
     */
    @Bean("directExchangeBindAQueue")
    public Binding directExchangeBindAQueue(@Qualifier(value = "directExchange") Exchange directExchange,
                                            @Qualifier(value = "queueAName") Queue queueAName) {
        return BindingBuilder.bind(queueAName).to(directExchange).with(ROUTE_KEY).noargs();
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-14 21:39
 * 测试生产者
 */
@Component
@RequiredArgsConstructor
public class TestProducer {
    //注入rabbitTemplate对象
    private final RabbitTemplate rabbitTemplate;
    /***
     * 生产者方法
     * @param msg 消息
     */
    public void producerSendMsg(MessageSendDTO msg) {
        //消息转换为JSON格式并转为字节数组
        byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8);
        //发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ROUTE_KEY, bytes);
    }
}
//----------------------------------------------------------------
 
 
/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-13 23:29
 * 这是监听队列A的消费者（A、B）
 */
@Slf4j
@Component
public class QueueConsumer {
    /***
     * 消费者A（监听）队列queueAName
     * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型
     * @param message 这个就类似我们原生的message
     * @param channel 这个就类似我们原生的channel
     */
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_A_NAME}) //只需要监听队列即可,多个则在{}里面逗号分割
    public void messageSimpleHandleA(@Payload String msgData, //这个是生产者发送的JSON消息
                                     Message message,
                                     Channel channel) throws InterruptedException, IOException {
        //获取到队列消息，因为发送是JSON格式，我们要解析对象格式
        MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class);
        //假设消费者A处理消息慢，每8秒处理一条
        Thread.sleep(8000);
        log.info("A：消息由消费者A消费：{}，并消费完成", msg);
    }
    /***
     * 消费者B（监听）队列queueAName
     * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型
     * @param message 这个就类似我们原生的message
     * @param channel 这个就类似我们原生的channel
     */
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_A_NAME}) //只需要监听队列即可,多个则在{}里面逗号分割
    public void messageSimpleHandleB(@Payload String msgData, //这个是生产者发送的JSON消息
                                     Message message,
                                     Channel channel) throws InterruptedException, IOException {
        //获取到队列消息，因为发送是JSON格式，我们要解析对象格式
        MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class);
        //假设消费者B处理消息快，每2秒处理一条
        Thread.sleep(2000);
        log.info("B：消息由消费者B消费：{}，并消费完成", msg);
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-14 21:37
 */
@Slf4j  //使用lombok自带的日志注解，具体实现是slf4j+log4j2
@RestController
@RequestMapping("/test")
@RequiredArgsConstructor
public class TestController {
    //使用SLF4J来获取Logger对象;(注意导包：import org.slf4j.Logger; import org.slf4j.LoggerFactory;)
    //Logger logger = LoggerFactory.getLogger(this.getClass());
    //注入生产者对象
    private final TestProducer testProducer;
    /***
     * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息
     * @param msg 请求消息
     * @return String
     */
    @PostMapping("/produce")
    public String msgSend(@RequestBody MessageSendDTO msg) {
        log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg);
        //循环发送消息
        for (int i = 97; i <= 106; i++) {
            MessageSendDTO build = MessageSendDTO.builder().msgID(i)
                    .msgType("testType")
                    .msgBody(msg.getMsgBody() + new String(new char[]{(char) i, (char) i})).build();
            //发送消息
            testProducer.producerSendMsg(build);
        }
        return "请求发送成功，并已接收";
    }
}

server:
  port: 8081
spring:
  ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties
  rabbitmq:
    host: 49.235.99.193
    port: 5672
    username: admin
    password: 123
    virtual-host: test

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-16 14:05
 * 扇出交换机配置
 */
@Configuration
public class RabbitMQConfig {
    //扇出交换机名称
    public static final String EXCHANGE_NAME = "fanoutDemo";
    //创建两个消息队列
    public static final String QUEUE_A = "queueA";
    public static final String QUEUE_B = "queueB";
    /***
     * 创建交换机信息
     * @return Exchange
     */
    @Bean("fanoutDemo")
    public Exchange fanoutDemo() {
        return ExchangeBuilder.fanoutExchange(EXCHANGE_NAME).durable(true).build();
    }
    /***
     * 创建队列A
     * @return Queue
     */
    @Bean("queueA")
    public Queue queueA() {
        return QueueBuilder.durable(QUEUE_A).build();
    }
    /***
     * 创建队列B
     * @return Queue
     */
    @Bean("queueB")
    public Queue queueB() {
        return QueueBuilder.durable(QUEUE_B).build();
    }
    /***
     * 队列A绑定到扇出交换机
     * @param fanoutDemo 交换机名称
     * @param queueA 队列A
     * @return Binding
     */
    @Bean("fanoutBindQueueA")
    public Binding fanoutBindQueueA(@Qualifier(value = "fanoutDemo") Exchange fanoutDemo,
                                    @Qualifier(value = "queueA") Queue queueA) {
        return BindingBuilder.bind(queueA).to(fanoutDemo).with("").noargs();
    }
    /***
     * 队列B绑定到扇出交换机
     * @param fanoutDemo 交换机名称
     * @param queueB 队列B
     * @return Binding
     */
    @Bean("fanoutBindQueueB")
    public Binding fanoutBindQueueB(@Qualifier(value = "fanoutDemo") Exchange fanoutDemo,
                                    @Qualifier(value = "queueB") Queue queueB) {
        return BindingBuilder.bind(queueB).to(fanoutDemo).with("").noargs();
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-16 15:01
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class TestProducer {
    private final RabbitTemplate rabbitTemplate;
    /***
     * 生产者方法
     * @param msg 消息
     */
    public void producerSendMsg(MessageSendDTO msg) {
        //消息转换为JSON格式并转为字节数组
        byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8);
        //发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, "", bytes);
        log.info("生产者发送信息完成，已经交由给交换机.....");
    }
}
//======================================
/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-16 15:04
 * 消费者信息
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class QueueConsumer {
    /***
     * 消费者A
     */
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_A})
    public void testConsumerA(@Payload String msgData, //这个是生产者发送的JSON消息
                              Message message,
                              Channel channel) {
        log.info("接收到队列A信息......；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class));
    }
    /***
     * 消费者B
     */
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_B})
    public void testConsumerB(@Payload String msgData, //这个是生产者发送的JSON消息
                              Message message,
                              Channel channel) {
        //注：若消费失败（报错）会自动手动不确认，并且把消息放到队列中，然后又被这个队列消费，最终死循环
        int a = 1 / 0;
        log.info("接收到队列B信息......；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class));
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-14 21:37
 */
@Slf4j
@RestController
@RequestMapping("/test")
@RequiredArgsConstructor
public class TestController {
    //注入生产者对象
    private final TestProducer testProducer;
    /***
     * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息
     * @param msg 请求消息
     * @return String
     */
    @PostMapping("/produce")
    public String msgSend(@RequestBody MessageSendDTO msg) {
        log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg);
        //发送消息
        testProducer.producerSendMsg(msg);
        return "请求发送成功，并已接收";
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-16 23:10
 * 主题交换机配置类
 */
@Configuration
public class RabbitMQConfig {
    //交换机名称
    public static final String TOPIC_EXCHANGE = "TopicExchange";
    //队列Q1名称
    public static final String Q1 = "Q1Queue";
    //队列Q2名称
    public static final String Q2 = "Q2Queue";
    //路由绑定关系 Routing Key
    public static final String Q1_KEY = "*.orange.*";
    //路由绑定关系 Routing Key 1
    public static final String Q2_KEY_A = "*.*.rabbit";
    //路由绑定关系 Routing Key 2
    public static final String Q2_KEY_B = "lazy.#";
    /***
     * 主题交换机
     * @return Exchange
     */
    @Bean("topicExchange")
    public Exchange topicExchange() {
        return ExchangeBuilder.topicExchange(TOPIC_EXCHANGE).durable(true).build();
    }
    /***
     * 队列1信息
     * @return Queue
     */
    @Bean("q1Queue")
    public Queue q1Queue() {
        return QueueBuilder.durable(Q1).build();
    }
    /***
     * 队列2信息
     * @return Queue
     */
    @Bean("q2Queue")
    public Queue q2Queue() {
        return QueueBuilder.durable(Q2).build();
    }
    /***
     * 绑定关系，Q1Queue队列绑定的匹配路由为*.orange.*
     * @param topicExchange 交换机
     * @param q1Queue 队列1
     * @return Binding
     */
    @Bean("bindingA")
    public Binding bindingA(@Qualifier("topicExchange") Exchange topicExchange,
                            @Qualifier("q1Queue") Queue q1Queue) {
        return BindingBuilder.bind(q1Queue).to(topicExchange).with(Q1_KEY).noargs();
    }
    /***
     * 绑定关系，Q2Queue队列绑定的匹配路由为*.*.rabbit
     * @param topicExchange 交换机
     * @param q2Queue 队列2
     * @return Binding
     */
    @Bean("bindingB1")
    public Binding bindingB1(@Qualifier("topicExchange") Exchange topicExchange,
                             @Qualifier("q2Queue") Queue q2Queue) {
        return BindingBuilder.bind(q2Queue).to(topicExchange).with(Q2_KEY_A).noargs();
    }
    /***
     * 绑定关系，Q2Queue队列绑定的匹配路由为lazy.#
     * @param topicExchange 交换机
     * @param q2Queue 队列2
     * @return Binding
     */
    @Bean("bindingB2")
    public Binding bindingB2(@Qualifier("topicExchange") Exchange topicExchange,
                             @Qualifier("q2Queue") Queue q2Queue) {
        return BindingBuilder.bind(q2Queue).to(topicExchange).with(Q2_KEY_B).noargs();
    }
}

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-16 15:01
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class TestProducer {
    private final RabbitTemplate rabbitTemplate;
    /***
     * 生产者方法
     */
    public void producerSendMsg() {
        //消息任务准备
        HashMap<String, String> sendMsg = new HashMap<>();
        sendMsg.put("quick.orange.rabbit", "被队列 Q1 Q2 接收到");
        sendMsg.put("lazy.orange.elephant", "被队列 Q1 Q2 接收到");
        sendMsg.put("quick.orange.fox", "被队列 Q1 接收到");
        sendMsg.put("lazy.brown.fox", "被队列 Q2 接收到");
        sendMsg.put("lazy.pink.rabbit", "虽然满足两个绑定规则但两个规则都是在Q2队列，所有只要Q2接收一次");
        sendMsg.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃");
        sendMsg.put("quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃");
        sendMsg.put("lazy.orange.male.rabbit", "是四个单词但匹配 Q2");
        //循环发送消息任务
        for (Map.Entry<String, String> msg : sendMsg.entrySet()) {
            String routKey = msg.getKey();  //主题路由key
            String message = msg.getValue();//消息任务
            //创建对象
            MessageSendDTO build = MessageSendDTO.builder().msgBody("基本信息：" + message + " 路由信息：" + routKey).build();
            //消息转换为JSON格式并转为字节数组
            byte[] bytes = JSONObject.toJSONString(build).getBytes(StandardCharsets.UTF_8);
            //发送消息
            rabbitTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, routKey, bytes);
        }
        log.info("生产者发送信息完成，已经交由给交换机.....");
    }
}
//+++++++++++++++++++++++++++++++++++++++
/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2023-04-16 15:04
 * 消费者信息
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class QueueConsumer {
    /***
     * 消费者1
     */
    @RabbitListener(queues = {RabbitMQConfig.Q1})
    public void testConsumerA(@Payload String msgData, //这个是生产者发送的JSON消息
                              Message message,
                              Channel channel) {
        log.info("接收到队列1信息；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class));
    }
    /***
     * 消费者2
     */
    @RabbitListener(queues = {RabbitMQConfig.Q2})
    public void testConsumerB(@Payload String msgData, //这个是生产者发送的JSON消息
                              Message message,
                              Channel channel) {
        log.info("接收到队列2信息......；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class));
    }
}