RabbitMQ的五种模型_rabbitmq五种消息模型

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

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.*;
import java.util.HashMap;
import java.util.Map;
 
/**
 * 简单队列模式（生产者）
 */
public class SimpleModeProducer {
 
    private static String QUEUE_NAME = "SIMPLE_MODE";
 
    public static void main(String[] args) throws Exception {
        Map map =new HashMap();
        map.put("text", "测试内容");
 
        //1.连接远程rabbit-server服务器
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        //2.创建一个连接
        Connection connection = factory.newConnection();
        //3.创建一个管道
        Channel channel = connection.createChannel();
        //4.定义创建一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //5.发送消息(注意发送和接受段相同字符集否则出现乱码)
        channel.basicPublish("", QUEUE_NAME, null, ser(map));
        System.out.println("生产成功！！");
        channel.close();
        connection.close();
    }
 
    /**
     * 将对象序列化为字节数组
     */
    private static byte[] ser(Object obj) throws IOException{
        ByteArrayOutputStream baos =new ByteArrayOutputStream();
        ObjectOutputStream oos =new ObjectOutputStream(baos);
        oos.writeObject(obj);
        return baos.toByteArray();
    }
}

import com.rabbitmq.client.*;
import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;
 
/**
 * 简单队列模式（消费者）
 */
public class SimpleModeConsumer {
 
    private static String QUEUE_NAME = "SIMPLE_MODE";
 
    public static void main(String[] args)throws Exception {
        //连接远程rabbit-server服务器
        ConnectionFactory factory = new ConnectionFactory();
        //MQ的服务ip
        factory.setHost("127.0.0.1");
        //端口
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        //创建一个连接
        Connection connection = factory.newConnection();
        //创建一个管道
        Channel channel = connection.createChannel();
        //创建一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义回调抓取消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
                try {
                    Map map = (Map) dser(body);
                    System.out.println(map.get("text").toString());
                    System.out.println("接收成功");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
 
    /**
     * 	反序列化
     */
    private static Object dser(byte[] src) throws Exception {
        // 从字节数组读取数据
        ByteArrayInputStream bis = new ByteArrayInputStream(src);
        // 把字节数组反序列化成对象
        ObjectInputStream ois = new ObjectInputStream(bis);
        return ois.readObject();
    }
}