【几种 mq 的区别及使用案例代码】-蒲公英云

【几种 mq 的区别及使用案例代码】

消息队列（Message Queue，MQ）是一种常见的异步通信方式，常用于解耦应用组件、提高系统可用性和扩展性、降低系统耦合度等。常见的消息队列有以下几种：

RabbitMQ RabbitMQ 是一款使用 Erlang 语言编写的 AMQP 消息队列，具有性能高、可扩展性好、支持多种编程语言、可靠性强等优点。RabbitMQ 的主要应用场景包括：异步处理、解耦应用、实现RPC、实现发布/订阅模式等。
ActiveMQ ActiveMQ 是一个基于 JMS 规范的消息队列，具有支持多种协议、支持多种语言、高可用性、可扩展性等特点。ActiveMQ 主要应用场景包括：解耦应用、异步通信、分布式系统、大数据处理等。
Kafka Kafka 是一款高性能、分布式、可扩展的消息队列，具有支持多种协议、支持多种语言、可靠性好、高吞吐量等特点。Kafka 主要应用场景包括：大数据处理、实时数据流处理、日志处理等。
RocketMQ RocketMQ 是一款使用 Java 语言编写的分布式消息队列，具有高可用性、高吞吐量、多种消息模式等特点。RocketMQ 主要应用场景包括：异步通信、流式计算、微服务等。
Redis Redis 是一个高性能的 key-value 存储系统，可以通过 List、Set、Hash 等数据结构实现消息队列的功能。Redis 主要应用场景包括：实时数据处理、消息推送、日志处理等。

以上几种消息队列在应用场景和特点上都有所不同，具体选择哪一种消息队列要根据业务需求和系统特点进行选择。在实际应用中，也可以使用多种消息队列协同工作，以达到更好的效果。
以下是几种 MQ 在 Java 中的使用案例代码：

RabbitMQ 生产者代码：

@Component
public class RabbitMQProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("my-exchange", "my-routing-key", message);
    }
}

消费者代码：

@Component
public class RabbitMQConsumer {
    @RabbitListener(queues = "my-queue")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

ActiveMQ 生产者代码：

@Component
public class ActiveMQProducer {
    @Autowired
    private JmsTemplate jmsTemplate;
    public void sendMessage(String message) {
        jmsTemplate.convertAndSend("my-destination", message);
    }
}

消费者代码：

@Component
public class ActiveMQConsumer implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            try {
                String messageContent = textMessage.getText();
                System.out.println("Received message: " + messageContent);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

Kafka 生产者代码：

@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendMessage(String message) {
        kafkaTemplate.send("my-topic", message);
    }
}

消费者代码：

@Component
public class KafkaConsumer {
    @KafkaListener(topics = "my-topic")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

RocketMQ 生产者代码：

@Component
public class RocketMQProducer {
    @Autowired
    private DefaultMQProducer producer;
    public void sendMessage(String message) throws Exception {
        Message mqMessage = new Message("my-topic", "my-tag", message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult result = producer.send(mqMessage);
        System.out.println("Message ID: " + result.getMsgId());
    }
}

消费者代码：

@Component
public class RocketMQConsumer implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        for (MessageExt message : messages) {
            try {
                String messageContent = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
                System.out.println("Received message: " + messageContent);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

Redis 生产者代码：

@Component
public class RedisProducer {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    public void sendMessage(String message) {
        redisTemplate.opsForList().leftPush("my-list", message);
    }
}

消费者代码：

@Component
public class RedisConsumer {
    @Scheduled(fixedDelay = 1000)
    public void receiveMessage() {
        String message = redisTemplate.opsForList().rightPop("my-list");
        if (message != null) {
            System.out.println("Received message: " + message);
        }
    }
}