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

时间: 2024-11-05 06:08:18

RabbitMQ/JAVA 客户端测试(补:利用线程)的相关文章

淘宝分布式 key/value 存储引擎Tair安装部署过程及Java客户端测试一例

目录 1. 简介 2. 安装步骤及问题小记 3. 部署配置 4. Java客户端测试 5. 参考资料 声明 1. 下面的安装部署基于Linux系统环境:centos 6(64位),其它Linux版本可能有所差异. 2. 网上有人说tair安装失败可能是因为gcc版本问题,高版本的gcc可能不支持某些特性导致安装失败,经过实验证明,该说法是错误的,tair安装失败有各种可能的原因但绝对与gcc版本无关,比如我的gcc开始版本为4.4.7,后来tair安装失败,我重新编译低版本的gcc(gcc4.1

docker安装fastdfs与java客户端测试

一.docker 安装FastDFS 1.拉取镜像 docker pull morunchang/fastdfs 2.创建并启动tracker容器 docker run -d --name=tracker -v /home/fastdfs_docker/fdfs/tracker:/data/fast_data --privileged=true --net=host morunchang/fastdfs sh tracker.sh 3.创建并启动storage容器.此处只做单机版测试 注意:由于

HBase的java客户端测试(一)---DDL操作

测试准备 [首先同步时间:] for node in CloudDeskTop master01 master02 slave01 slave02 slave03;do ssh $node "date -s '2017-12-30 21:32:30'";done [slave各节点启动zookeeper集群:] cd /software/zookeeper-3.4.10/bin/ && ./zkServer.sh start && cd - &&

HBase的java客户端测试(二)---DML操作

测试准备 [首先同步时间:] for node in CloudDeskTop master01 master02 slave01 slave02 slave03;do ssh $node "date -s '2017-12-30 21:32:30'";done [slave各节点启动zookeeper集群:] cd /software/zookeeper-3.4.10/bin/ && ./zkServer.sh start && cd - &&

RabbitMQ(3) Java客户端使用

RabbitMQ针对不同的开发语言(java,python,c/++,Go等等),提供了丰富对客户端,方便使用.就Java而言,可供使用的客户端有RabbitMQ Java client. RabbitMQ JMS client.apache的camel-rabbitmq.以及Banyan等.在Spring中,也可以使用Spring AMQP.Spring Cloud Data Flow方便对集成RabbitMQ. 实际开发使用中,RabbitMQ Java client和Spring AMQP

Tedis:淘宝的Redis的Java客户端开发包

http://www.open-open.com/lib/view/open1389880631976.html Tedis Tedis是另一个redis的java客户端 Tedis的目标是打造一个可在生产环境直接使用的高可用Redis解决方案.参见https://github.com/alibaba/tb_tddl Feature 高可用,Tedis使用多写随机读做HA确保redis的高可用 高性能,使用特殊的线程模型,使redis的性能不限制在客户端 多种使用方式,如果你只有一个redis实

Java 理论和实践:线程池和工作队列

使用线程池以获取最佳资源利用率 Java 多线程编程论坛中最常见的一个问题就是各种版本的 "我怎么样才可以创建一个线程池?" 几乎在每个服务器应用里,都会出现关于线程池和工作队列的问题.本文中,Brian Goetz 就线程池原理.基本实现和调优技术.需要避开的一些常见误区等方面进行共享. 为何要用线程池? 有很多服务器应用,比如 Web 服务器,数据库服务器,文件服务器,或者邮件服务器,都会面对处理大量来自一些远程请求的小任务.请求可能会以很多种方式到达服务器,比如通过一种网络协议(

java多线程总结五:线程池的原理及实现

1.线程池简介:     多线程技术主要解决处理器单元内多个线程执行的问题,它可以显著减少处理器单元的闲置时间,增加处理器单元的吞吐能力.        假设一个服务器完成一项任务所需时间为:T1 创建线程时间,T2 在线程中执行任务的时间,T3 销毁线程时间.    如果:T1 + T3 远大于 T2,则可以采用线程池,以提高服务器性能.                 一个线程池包括以下四个基本组成部分:                 1.线程池管理器(ThreadPool):用于创建并管

ZPush--基于netty4实现的苹果通知推送服务(APNs)Java客户端

简单说下实现苹果通知推送服务(APNs)客户端的一些要注意的地方: 使用长连接: sanbox服务器是没用的,调试时直接用"gateway.push.apple.com"域名: 对于错误的Notification,苹果会回应一个Error response,里面有个identifier,在这个identifier之后的Notification全都失败: 因此发送者要缓存已经发送的Notification,最好设置Notification identifier为增长的整数序列,当收到Er