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

朴灿烈づ我的快乐病毒、 2024-03-26 14:50 29阅读 0赞

消息队列（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);
            }
        }
    }