<!-- springboot集成rabbitMQ -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

  #rabbitmq配置
spring:
  rabbitmq:
    host: ip地址
    port: 5672
    username: 賬號
    password: 密碼
    virtual-host: /

package com.ckm.ball.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
//rabbitMQ綁定交換機(jī) / 隊(duì)列
@Configuration
public class RabbitMQConfig {
    //========================================================RabbitMQ Queue========================================================//
    //創(chuàng)建fanout模式交換機(jī)
    @Bean
    public FanoutExchange fanoutExchangeProcess() {
        return new FanoutExchange("process-data-change-exchange", true, false);
    }
    //創(chuàng)建隊(duì)列
    @Bean
    public Queue processDataChangeQueue() {
        return new Queue("process-data-change-queue", true);
    }
    //將隊(duì)列綁定到交換機(jī)
    @Bean
    public Binding chatBindExchange() {
        return BindingBuilder.bind(processDataChangeQueue()).to(fanoutExchangeProcess());
    }
}

@Resource
private RabbitTemplate rabbitTemplate;
@GetMapping("/produceMessage")
@ApiOperation(value = "生產(chǎn)消息", tags = "測試接口")
public void updateTokenTime() {
	//生產(chǎn)消息，會到交換機(jī)，交換機(jī)下發(fā)給隊(duì)列，隊(duì)列監(jiān)聽到就會消費(fèi)，執(zhí)行業(yè)務(wù)邏輯
    rabbitTemplate.convertAndSend("process-data-change-exchange", "process-data-change-queue", "hhhhhhhhhhhhhh");
}

package com.ckm.ball.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import java.util.Date;
@Slf4j
@Component
public class RabbitMQDataSyncListenerProcess {
    //監(jiān)聽process-data-change-queue隊(duì)列 -> 消費(fèi)
    @RabbitListener(queues = "process-data-change-queue")
    public void orderDead(@Payload String productIdAndOrderId) {
        log.info("當(dāng)前時間：{},收到隊(duì)列信息：{}", new Date().toString(), productIdAndOrderId);
        //執(zhí)行你的業(yè)務(wù)邏輯
        for (int i = 0; i < 5; i++) {
            System.out.println("循環(huán)次數(shù): " + (i + 1));
            try {
                // 暫停 2000 毫秒（2 秒）
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                // 處理異常
                System.err.println("線程被中斷: " + e.getMessage());
            }
        }
    }
}

rabbitTemplate.convertAndSend("process-data-change-exchange", "process-data-change-queue", "hhhhhhhhhhhhhh");

rabbitTemplate.convertAndSend("process-data-change-exchange", "", "hhhhhhhhhhhhhh");

// 創(chuàng)建直連交換機(jī)
@Bean
public DirectExchange directExchange() {
    return new DirectExchange("direct-exchange", true, false);
}
// 創(chuàng)建隊(duì)列
@Bean
public Queue directQueue() {
    return new Queue("direct-queue", true);
}
// 將隊(duì)列綁定到直連交換機(jī)，同時指定路由鍵
@Bean
public Binding directBinding() {
    return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct-routing-key");
}

// 發(fā)送消息到直連交換機(jī)
rabbitTemplate.convertAndSend("direct-exchange", "direct-routing-key", "Your message here");

// 創(chuàng)建主題交換機(jī)
@Bean
public TopicExchange topicExchange() {
    return new TopicExchange("topic-exchange", true, false);
}
// 創(chuàng)建隊(duì)列
@Bean
public Queue topicQueue() {
    return new Queue("topic-queue", true);
}
// 將隊(duì)列綁定到主題交換機(jī)，同時指定路由鍵
@Bean
public Binding topicBinding() {
    return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.#");
}

// 發(fā)送消息到主題交換機(jī)
rabbitTemplate.convertAndSend("topic-exchange", "topic.routing.key", "Your message here");

// 創(chuàng)建扇出交換機(jī)
@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("fanout-exchange", true, false);
}
// 創(chuàng)建隊(duì)列
@Bean
public Queue fanoutQueue1() {
    return new Queue("fanout-queue-1", true);
}
@Bean
public Queue fanoutQueue2() {
    return new Queue("fanout-queue-2", true);
}
// 將隊(duì)列綁定到扇出交換機(jī)
@Bean
public Binding fanoutBinding1() {
    return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding2() {
    return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
}

// 發(fā)送消息到扇出交換機(jī)
rabbitTemplate.convertAndSend("fanout-exchange", "", "Your message here");

// 創(chuàng)建頭交換機(jī)
@Bean
public HeadersExchange headersExchange() {
    return new HeadersExchange("headers-exchange", true, false);
}
// 創(chuàng)建隊(duì)列
@Bean
public Queue headersQueue() {
    return new Queue("headers-queue", true);
}
// 將隊(duì)列綁定到頭交換機(jī)，同時指定頭屬性
@Bean
public Binding headersBinding() {
    Map<String, Object> headers = new HashMap<>();
    headers.put("format", "pdf");
    headers.put("type", "report");
    return BindingBuilder.bind(headersQueue())
            .to(headersExchange())
            .whereAll(headers)
            .match();
}

// 發(fā)送消息到頭交換機(jī)
MessageProperties messageProperties = new MessageProperties();
messageProperties.setHeader("format", "pdf");
messageProperties.setHeader("type", "report");
Message message = new Message("Your message here".getBytes(), messageProperties);
rabbitTemplate.send("headers-exchange", "", message);