Rabbitmq

brew install rabbitmq
#查看版本信息,及安装位置
brew info rabbitmq
#配置环境变量
open -e ~/.zshrc
#添加如下内容
export PATH=/opt/homebrew/Cellar/rabbitmq/3.12.13/sbin:$PATH

# 启动服务
sudo rabbitmq-server
# rabbitmq 后台启动命令
sudo rabbitmq-server -detached
# 关闭服务
sudo rabbitmqctl stop
# 查询rabbitmq的状态
sudo rabbitmqctl status
# 启动插件
sudo rabbitmq-plugins enable rabbitmq_management
# 查找rabbitmq在哪
find / -name 'rabbitmq' -type d
# rabbitmq内存限制
# 系数计算，假设机器内存32g,以下命令限制的最高内存为，32*0.06=1.92g
rabbitmqctl set_vm_memory_high_watermark 0.06
# rabbitmq添加消息日志记录
rabbitmq-plugins enable rabbitmq_tracing
# 查询rabbitmq的进程
ps -ef | grep rabbitmq
# 杀掉rabbitmq进程
ps -ef | grep rabbitmq | grep -v grep | awk '{print $2}' | xargs kill -9

sudo rabbitmq-plugins enable rabbitmq_tracing

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.9.0</version>
</dependency>

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class Producer {
    private final static String QUEUE_NAME = "queue_name";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "这是一条来自生产者的消息。";
            //发布消息basicPublish(交换机,路由关键字,其他参数,消息主体)
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println("生产者发送: '" + message + "'");
        }
    }
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer {
    private final static String QUEUE_NAME = "queue_name";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //必须设置队列名称
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println("等待接收消息");
        //RabbitMQ将会通过异步的方式推送消息，因此需要提供了一个回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" 消费者 接收到消息: '" + message + "'");
        };
        //basicConsume用于接收消息参数(队列名,autoAck,回调函数,CancelCallback)
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    }
}

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

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * rabbitmq配置类
 **/
@Configuration
public class DirectExchangeConfig {
    //队列1
    public static final String DIRECT_QUEUE = "directQueue";
    //队列2
    public static final String DIRECT_QUEUE2 = "directQueue2";
    //交换机,用来将路由和队列绑定,一个路由可以绑定多个队列,这样每次发送路由就会向多个队列发送消息
  //一个队列也可以绑定多个消费者,但是只有一个消费者能消费
    public static final String DIRECT_EXCHANGE = "directExchange";
    //路由,多个路由时
    public static final String DIRECT_ROUTING_KEY = "direct";
    //队列一
    @Bean
    public Queue directQueue() {
        return new Queue(DIRECT_QUEUE, true);
    }
    //队列二
    @Bean
    public Queue directQueue2() {
        return new Queue(DIRECT_QUEUE2, true);
    }
    //交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(DIRECT_EXCHANGE, true, false);
    }
    //将队列1和交换机绑定, 并设置路由
    @Bean
    public Binding bindingDirectExchange(Queue directQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);
    }
    //将队列2和交换机绑定, 并设置路由
    @Bean
    public Binding bindingDirectExchange2(Queue directQueue2, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue2).to(directExchange).with(DIRECT_ROUTING_KEY);
    }
}

spring:
    rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
 * @className: DirectQueueListener
 * @description: 直连交换机的监听器
 */
@Component
public class DirectQueueListener {
    /**
     * 尽管队列1设置了两个消费者，但是只有一个能够消费成功
     * 多次发送则轮训消费：
     * DirectReceiver消费者收到消息1  : 发送一条测试消息：direct
     * DirectReceiver消费者收到消息3  : 发送一条测试消息：direct
     * DirectReceiver消费者收到消息2  : 发送一条测试消息：direct
     * DirectReceiver消费者收到消息3  : 发送一条测试消息：direct
     *
     * 一个交换机可以绑定多个队列。如果通过路由key可以匹配到多个队列，消费的时候也只能有一个进行消费
     * @param testMessage
     */
    @RabbitHandler
    @RabbitListener(queues = DirectExchangeConfig.DIRECT_QUEUE)
    public void process(String testMessage) {
        System.out.println("DirectReceiver消费者收到消息1  : " + testMessage);
    }
    @RabbitHandler
    @RabbitListener(queues = DirectExchangeConfig.DIRECT_QUEUE)
    public void process2(String testMessage) {
        System.out.println("DirectReceiver消费者收到消息2  : " + testMessage);
    }
    /**
     * 处理消息后发送确认消息(确认后消息会从mq中移除)
     * @param message
     * @param channel
     */
    @RabbitHandler
    @RabbitListener(queues = DirectExchangeConfig.DIRECT_QUEUE2)
    public void process3(Message message, Channel channel) {
        try {
            String messageBody = new String(message.getBody(), "UTF-8");
            System.out.println("DirectReceiver消费者3收到消息  : " + messageBody);
            // 发送确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

@Controller
public class RabbitMqController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // http://127.0.0.1:8080/hellorabbitmq?msg=向rabbitmq发送的消息
    @RequestMapping("/hellorabbitmq")
    @ResponseBody
    public String hello(String msg) {
    //交换机,路由,消息体,然后加入到对应路由下面的队列
        rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE, DirectExchangeConfig.DIRECT_ROUTING_KEY, "发送一条测试消息：direct");
        return "客户端发送的消息:  " + msg;
    }
}

@Bean
    public Queue directQueue() {
        /**
         * 绑定死信交换机及路由key
         */
        Map<String, Object> args = new HashMap<>();
        // x-dead-letter-exchange：这里声明当前业务队列绑定的死信交换机
        //消息被拒绝、消息过期，或者队列达到其最大长度。消息会变成死信
        args.put("x-dead-letter-exchange", DEAD_TCP_DATA_DIRECT_EXCHANGE);
        // x-dead-letter-routing-key：这里声明当前业务队列的死信路由 key
        args.put("x-dead-letter-routing-key", DEAD_TCP_DATA_DIRECT_ROUTING);
        return QueueBuilder.durable(DIRECT_QUEUE).withArguments(args).build();
    }