从原理到实战，手把手教你在项目中使用RabbitMQ

brew update

brew install rabbitmq

# 启动方式1：后台启动
brew services start rabbitmq
# 启动方式2：当前窗口启动
cd /usr/local/Cellar/rabbitmq/3.8.19
rabbitmq-server

http://localhost:15672/

## 添加账号
./rabbitmqctl add_user admin admin
## 添加访问权限
./rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
## 设置超级权限
./rabbitmqctl set_user_tags admin administrator

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

/**
 * @author Louzai
 * @date 2023/5/10
 */
public class RabbitmqUtil {

    /**
     * 每个key都有自己的工厂
     */
    private static Map<String, ConnectionFactory> executors = new ConcurrentHashMap<>();

    /**
     * 初始化一个工厂
     *
     * @param host
     * @param port
     * @param username
     * @param passport
     * @param virtualhost
     * @return
     */
    public static ConnectionFactory init(String host,
                                  Integer port,
                                  String username,
                                  String passport,
                                  String virtualhost) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(passport);
        factory.setVirtualHost(virtualhost);
        return factory;
    }

    /**
     * 工厂单例，每个key都有属于自己的工厂
     *
     * @param key
     * @param host
     * @param port
     * @param username
     * @param passport
     * @param virtualhost
     * @return
     */
    public static ConnectionFactory getOrInitConnectionFactory(String key,
                                                               String host,
                                                               Integer port,
                                                               String username,
                                                               String passport,
                                                               String virtualhost) {
        ConnectionFactory connectionFactory = executors.get(key);
        if (null == connectionFactory) {
            synchronized (RabbitmqUtil.class) {
                connectionFactory = executors.get(key);
                if (null == connectionFactory) {
                    connectionFactory = init(host, port, username, passport, virtualhost);
                    executors.put(key, connectionFactory);
                }
            }
        }
        return connectionFactory;
    }
}

/**
 * @author Louzai
 * @date 2023/5/10
 */
@Component
public class RabbitmqClient {

    @Autowired
    private RabbitmqProperties rabbitmqProperties;

    /**
     * 创建一个工厂
     * @param key
     * @return
     */
    public ConnectionFactory getConnectionFactory(String key) {
        String host = rabbitmqProperties.getHost();
        Integer port = rabbitmqProperties.getPort();
        String userName = rabbitmqProperties.getUsername();
        String password = rabbitmqProperties.getPassport();
        String virtualhost = rabbitmqProperties.getVirtualhost();
        return RabbitmqUtil.getOrInitConnectionFactory(key, host, port, userName,password, virtualhost);
    }
}

@Component
public class RabbitmqServiceImpl implements RabbitmqService {

    @Autowired
    private RabbitmqClient rabbitmqClient;

    @Autowired
    private NotifyService notifyService;

    @Override
    public void publishMsg(String exchange,
                           BuiltinExchangeType exchangeType,
                           String toutingKey,
                           String message) throws IOException, TimeoutException {
        ConnectionFactory factory = rabbitmqClient.getConnectionFactory(toutingKey);

        // TODO: 这种并发量起不来，需要改造成连接池

        //创建连接
        Connection connection = factory.newConnection();
        //创建消息通道
        Channel channel = connection.createChannel();

        // 声明exchange中的消息为可持久化，不自动删除
        channel.exchangeDeclare(exchange, exchangeType, true, false, null);

        // 发布消息
        channel.basicPublish(exchange, toutingKey, null, message.getBytes());

        System.out.println("Publish msg:" + message);
        channel.close();
        connection.close();
    }

    @Override
    public void consumerMsg(String exchange,
                            String queue,
                            String routingKey) throws IOException, TimeoutException {
        ConnectionFactory factory = rabbitmqClient.getConnectionFactory(routingKey);

        // TODO: 这种并发量起不来，需要改造成连接池

        //创建连接
        Connection connection = factory.newConnection();
        //创建消息信道
        final Channel channel = connection.createChannel();
        //消息队列
        channel.queueDeclare(queue, true, false, false, null);
        //绑定队列到交换机
        channel.queueBind(queue, exchange, routingKey);

        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Consumer msg:" + message);

                // 获取Rabbitmq消息，并保存到DB
                // 说明：这里仅作为示例，如果有多种类型的消息，可以根据消息判定，简单的用 if...else 处理，复杂的用工厂 + 策略模式
                notifyService.saveArticleNotify(JsonUtil.toObj(message, UserFootDO.class), NotifyTypeEnum.PRAISE);

                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 取消自动ack
        channel.basicConsume(queue, false, consumer);
    }

    @Override
    public void processConsumerMsg() {
        System.out.println("Begin to processConsumerMsg.");

        Integer stepTotal = 1;
        Integer step = 0;

        // TODO: 这种方式非常 Low，后续会改造成阻塞 I/O 模式
        while (true) {
            step ++;
            try {
                System.out.println("processConsumerMsg cycle.");
                consumerMsg(CommonConstants.EXCHANGE_NAME_DIRECT, CommonConstants.QUERE_NAME_PRAISE,
                        CommonConstants.QUERE_KEY_PRAISE);
                if (step.equals(stepTotal)) {
                    Thread.sleep(10000);
                    step = 0;
                }
            } catch (Exception e) {

            }
        }
    }
}

// 点赞消息走 RabbitMQ，其它走 Java 内置消息机制
if (notifyType.equals(NotifyTypeEnum.PRAISE) && rabbitmqProperties.getSwitchFlag()) {
    rabbitmqService.publishMsg(
            CommonConstants.EXCHANGE_NAME_DIRECT,
            BuiltinExchangeType.DIRECT,
            CommonConstants.QUERE_KEY_PRAISE,
            JsonUtil.toStr(foot));
} else {
    Optional.ofNullable(notifyType).ifPresent(notify -> SpringUtil.publishEvent(new NotifyMsgEvent<>(this, notify, foot)));
}

@Override
public void run(ApplicationArguments args) {
    // 设置类型转换, 主要用于mybatis读取varchar/json类型数据据，并写入到json格式的实体Entity中
    JacksonTypeHandler.setObjectMapper(new ObjectMapper());
    // 应用启动之后执行
    GlobalViewConfig config = SpringUtil.getBean(GlobalViewConfig.class);
    if (webPort != null) {
        config.setHost("http://127.0.0.1:" + webPort);
    }
    // 启动 RabbitMQ 进行消费
    if (rabbitmqProperties.getSwitchFlag()) {
        taskExecutor.execute(() -> rabbitmqService.processConsumerMsg());
    }
    log.info("启动成功，点击进入首页: {}", config.getHost());
}