zoukankan      html  css  js  c++  java
  • RabbitMQ/JAVA 客户端测试(补:利用线程)

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

    我是将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

  • 相关阅读:
    Python读写ini文件
    MySQL 让主键id 从1开始自增
    python 找字符串中所有包含字符的下标
    centos7防火墙命令
    如何将npm升级到最新版本
    将 Npm 的源替换成淘宝的源
    MySQL 时间格式化/ MySQL DATE_FORMAT
    Python中crypto模块进行AES加密和解密
    windows环境下python3安装Crypto
    Nginx+PHPSTORM+Xdebug 配置
  • 原文地址:https://www.cnblogs.com/tinmh/p/6062197.html
Copyright © 2011-2022 走看看