RabbitMQ/JAVA 客户端测试（补：利用线程）

Posted 2020-08-19 itnnn

上次进行了简单的连接测试。这次主要进行一下小小的补充。利用线程将生产者消费者代码合到一个文件中。

我是将Recv.java（消费者）文件放在一个线程里添加到Send.java（生产者）中。

代码如下：

package MYRABBITMQ;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Scanner;

import com.rabbitmq.client.Channel;
class RecvThread extends Thread{
    RecvThread(String name){
        super(name);//调用父类带参数的构造方法
    }
    public void run(){
         final String QUEUE_NAME = "hello";
                 ConnectionFactory factory = new ConnectionFactory();  
                    factory.setHost("123.206.216.31");  
                    factory.setPort(5672);
                    factory.setUsername("admin");
                    factory.setPassword("admin");
                    // 打开连接和创建频道，与发送端一样  
                    Connection connection = null;
                    try {
                        connection = factory.newConnection();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }  
                    Channel channel = null;
                    try {
                        channel = connection.createChannel();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }  
              
                    // 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。  
                    try {
                        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }  
                    // 创建队列消费者  

                    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
                    // 指定消费队列  
                    // 指定消费队列 
                    QueueingConsumer consumer = new QueueingConsumer(channel);  
                    try {
                        channel.basicConsume(QUEUE_NAME, true, consumer);
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }  
                    while(true){  
                        QueueingConsumer.Delivery delivery = null;
                        try {
                            delivery = consumer.nextDelivery();
                        } catch (ShutdownSignalException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        } catch (ConsumerCancelledException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }  
                        String message = new String(delivery.getBody());  
                        System.out.println(" [x] Received ‘" + message + "‘");  
                    }  

            }

    }

public class Send {
     private final static String QUEUE_NAME = "hello";
      public static void main(String[] args) throws java.io.IOException {  
            /** 
             * 创建连接连接到MabbitMQ 
             */  
            // 设置MabbitMQ所在主机ip或者主机名

            ConnectionFactory factory = new ConnectionFactory();  
            factory.setHost("123.206.216.31");  
            factory.setPort(5672);
            System.out.println("input your username:");
            Scanner un=new Scanner(System.in);
            String username=null;
            username=un.next(); 
            System.out.println("input your password::");
            Scanner up=new Scanner(System.in);
            String userpassword=null;
            userpassword=up.nextLine(); 
            factory.setUsername(username);
            factory.setPassword(userpassword);
            // 创建一个连接  
            Connection connection = factory.newConnection();  
            // 创建一个频道  
            Channel channel = connection.createChannel();  
            // 指定一个队列  
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
            // 发送的消息 
            System.out.println("input what you wanna send:");
            Scanner s=new Scanner(System.in);
            String message=null;
            message=s.nextLine(); 
            // 往队列中发出一条消息  
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());  
            System.out.println(" [x] Sent ‘" + message + "‘");  
            // 关闭频道和连接  
            channel.close();  
            connection.close();  
            RecvThread rthread= new RecvThread("test1");
            rthread.start();
        }  

}

 另外，这里对于rabbitmq中的典型概念进行一个梳理：

刚刚接触RabbitMQ的时候，可能会被扑面而来的一堆概念搞的有些糊涂，又是 Exchange（交换器）、又是Queue（队列）、又是Routing_Key（路由）；还有什么 Connection（连接）、Channel（管道）。而且各个官网的例子当中，有些概念会出现、有些又不会出现，也没有特别的讲清楚，所以我们还是要先把概念和流程搞搞清楚。

既然我们用RabbitMQ来处理消息，那么我们就要清楚这个消息到底是什么。

（例1）：从实体的角度，我们可以把消息（Message）看成是个货物：我们从某宝、某东上买的货物，这个货物由厂家生产出来、通过卖家、快递送给我们，其中：货物就是“消息”，厂家就是生产者（publisher）、买家就是消费者（consumer）、某宝等平台就是交换器（Exchange）、物流快递就是队列（queue）。

（例2）：从虚拟的角度，我们也可以把消息 看成是个订单，这个时候，我们就是订单（Message）的生产者、而商家就成了订单的消费者，而这个时候的队列就不是快递而是运行在电商平台（Exchange）上的信息流了。

把大象放进冰箱要几步？ 
答：3步,第一步把冰箱门打开,第二步把大象放进去,第三步把冰箱门关上。

那么我们要在电商平台上下一个订单需要哪几步? 
第一步：打开电脑 
第二步：打开浏览器 
第三步：打开某东网站 
第四步：选定商品 
第五步：选择付款方式（某东支付 or 白条支付 or 某信支付） 
第六步：确认付款 
第七步：关闭浏览器 
第八步：关闭电脑

我们把消费的过程和消息队列的生命周期对应一下，会发现有一定相似性 
第一步：打开电脑 （连接到指定的RabbitMQ服务器 ，建立connection） 
第二步：打开浏览器 （获取信道、即open channel） 
第三步：打开某东网站 （声明交换器、即exchange_declare） 
第四步：选定商品 （声明队列、即 queue_declare） 
第五步：选择付款方式 （把队列和交换器绑定起来、即 queue_bind） 
第六步：确认付款 （消费信息、即 basic_consume ） 
第七步：关闭浏览器 （关闭信道、即 close channel） 
第八步：关闭电脑 （关闭连接、即 close connection）

一个完整的消息过程要包含上述八步，但是在打开电脑之后，其实你可以选择的事情有很多：你可以打开不止一个浏览器（channel）、也可以在不同的浏览器里打开很多个电商网站（exchange）、更不用说你在网站上可以任意的选择商品（queue）、以及选择喜欢的支付方式（queue_bind）。也就是说，connection只有一个、但是channel、exchange、exchange_type、queue、routing_key都可以根据一定的规则以及需求来组织。

概念部分转载自：http://m.blog.csdn.net/article/details?id=51039938