rabbitMq一收一发

ゞ浴缸里的玫瑰 2022-06-11 02:10 149阅读 0赞

1.生产者(producer)  
Producing  
发送消息的程序就叫做生产者。  
2.队列  
消息通过你的应用程序和RabbitMQ进行传输，它们能够只存储在一个队列（queue）中。队列（queue）没有任何限制，你要存储多少消息都可以——基本上是一个无限的缓冲。多个生产者（producers）能够把消息发送给同一个队列，同样，多个消费者（consumers）也能够从一个队列（queue）中获取数据。队列可以用下图标识：  
Queue  
3.一个消费者（consumer）就是一个等待获取消息的程序。我们把它画作”C”  
Consuming  
下面实现一个一发一收的demo:  
在此之前首先记录一下默认的端口和相关的作用:  
4369 (epmd), 25672 (Erlang distribution)  
5672, 5671 (AMQP 0-9-1 without and with TLS)  
15672 (if management plugin is enabled)  
61613, 61614 (if STOMP is enabled)  
1883, 8883 (if MQTT is enabled)  
接收队列会不断阻塞在那边监听发送端的消息，发送端如果发送了数据，立马会关闭连接。  
  
  
这里我们需要使用到5672端口，同时接收端和服务端的队列  
名字要一致，具体代码如下:  
接收端:

public class Receive {
        //消息队列名称
        private final static String QUEUE_NAME = "test1";
    
    
        public static void main(String[] args) throws Exception {
            //打开连接和创建频道，与发送端一样
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            //指定用户 密码
            factory.setUsername("lsj");
            factory.setPassword("123456");
            //指定端口
            factory.setPort(AMQP.PROTOCOL.PORT);
            //创建一个连接
            Connection connection = factory.newConnection();
            //创建一个频道
            Channel channel = connection.createChannel();
            //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    
            //创建队列消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            //指定消费队列
            channel.basicConsume(QUEUE_NAME, true, consumer);
            while (true)
            {
                //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("Received Message：'" + message + "'");
            }
        }
    }

发送端：

public class Send {
        //消息队列名称
        private final static String QUEUE_NAME = "test1";
    
    
        public static void main(String[] args) throws Exception {
            /**
             * 创建连接连接到MabbitMQ
             */
            ConnectionFactory factory = new ConnectionFactory();
            //设置MabbitMQ所在主机ip或者主机名
            factory.setHost("127.0.0.1");
            //指定用户密码
            factory.setUsername("lsj");
            factory.setPassword("123456");
            //指定端口
            factory.setPort(AMQP.PROTOCOL.PORT);
            //创建一个连接
            Connection connection = factory.newConnection();
            //创建一个频道
            Channel channel = connection.createChannel();
            //指定一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //发送的消息
            String message = "天王盖地虎~~";
            //往队列中发出一条消息
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("Sent Message：'" + message + "'");
            //关闭频道和连接
            channel.close();
            connection.close();
        }
    }