Java实现RabbitMQ(Simple)简单模式

Java实现RabbitMQ(Simple)简单模式 RabbitMQ

        RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

使用

        生产者

                这里我连接的是IP为192.168.111.130服务器上的RabbitMQ

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;


public class Producer {
    public static void main(String[] args) {
        //1.创建连接工程
        ConnectionFactory connectionFactory =new ConnectionFactory();
        connectionFactory.setHost("192.168.111.130");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //发在根节点上
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2.创建连接Connection
            connection = connectionFactory.newConnection("生产者");
            //3.通过连接获取通道Channel
            channel = connection.createChannel();
            //4.通过创建交换机，声明队列，绑定关系，路由Key,发送消息和接收消息
            String queueName = "queue1";
            
            channel.queueDeclare(queueName,false,false,false,null);
            //5.准备消息内容
            String message = "HelloWorld";
            //6.发送消息给队列queue
            channel.basicPublish("",queueName,null,message.getBytes());
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally {
            //7.关闭连接
            if(channel!=null && channel.isOpen()){
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            //8.关闭通道
            if(connection!=null && connection.isOpen()){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}

连接成功

 界面上可以看到

 连续运行了四次

 消费者

        这里我连接的是IP为192.168.111.130服务器上的RabbitMQ

import com.rabbitmq.client.*;
import sun.plugin2.message.Message;
import java.io.IOException;
import java.util.concurrent.TimeoutException;


public class Consumer {
    public static void main(String[] args) {
        //1.创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.111.130");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //发在根节点上
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2.创建连接Connection
            connection = connectionFactory.newConnection("生产者");
            //3.通过连接获取通道Channel
            channel = connection.createChannel();
            //4.通过创建交换机，声明队列，绑定关系，路由Key,发送消息和接收消息
            //接收消息
            channel.basicConsume("queue1", true, new DeliverCallback() {
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println("收到消息是" + new String(delivery.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                public void handle(String s) throws IOException {
                    System.out.println("接受失败了......");
                }
            });
            System.out.println("开始接收消息");
            System.in.read();
        } catch (
                IOException e)
        {
            e.printStackTrace();
        } catch (
                TimeoutException e)
        {
            e.printStackTrace();
        } finally
        {
            //7.关闭连接
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            //8.关闭通道
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

四条消息消费成功

 当消费者者程序关闭则非持久化队列删除，需要重启下rabbitmq服务