1.搭建并配置好rabbitmq服务

https://blog.csdn.net/qq_36932624/article/details/103245045

2.生产消息并发送

public class SendMessage {
 
    private final static String QUEUE_NAME = "hello";
 
    public static void main(String[] argv) throws Exception {
 
        //定义一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务器地址
        factory.setHost("127.0.0.1");
        //设置端口号
        factory.setPort(5672);
        //设置vhost
        factory.setVirtualHost("/aliyun");
 
        factory.setConnectionTimeout(10000);
        factory.setUsername("ceshi");
        factory.setPassword("1234");
        //获取一个连接
        Connection connection = factory.newConnection();
        //从连接中获取一个通道
        Channel channel = connection.createChannel();
        //声明一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "Hello RabbitMQ!";
        //发送消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
    }
}

3.消费消息

public class ReceiveMessage {
 
    private final static String QUEUE_NAME = "hello";
 
    public static void main(String[] args) throws Exception{
        //定义一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务器地址
        factory.setHost("127.0.0.1");
        //设置端口号
        factory.setPort(5672);
        //设置vhost
        factory.setVirtualHost("/aliyun");
 
        factory.setConnectionTimeout(10000);
        factory.setUsername("ceshi");
        factory.setPassword("1234");
 
        //获取一个连接
        Connection connection = factory.newConnection();
        //获取一个通道
        Channel channel = connection.createChannel();
        //定义队列的消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            //获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body,"utf-8");
                System.out.println(msg);
            }
        };
        //监听队列
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
    }
}