RabbitMQ 学习开发笔记

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel,这三个都是RabbitMQ对外提供的API中最基本的对象。不管是服务器端还是客户端都会首先创建这三类对象。
       ConnectionFactory为Connection的制造工厂。

Connection是与RabbitMQ服务器的socket链接,它封装了socket协议及身份验证相关部分逻辑。

Channel是我们与RabbitMQ打交道的最重要的一个接口,大部分的业务操作是在Channel这个接口中完成的,包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Queue

Queue(队列)是RabbitMQ的内部对象,用于存储消息

  RabbitMQ中的消息都只能存储在Queue中,生产者生产消息并最终投递到Queue中,消费者可以从Queue中获取消息并消费。

  队列是有Channel声明的,而且这个操作是幂等的。同名的队列多次声明也只会创建一次。我们发送消息就是想这个声明的队列里发送消息

  工作队列的主要思想是不用等待资源密集型的任务处理完成,

为了确保消息或者任务不会丢失,rabbitmq 支持消息确信 ACK。ACK机制是消费者端从rabbitmq收到消息并处理完成后,反馈给rabbitmq,rabbitmq收到反馈信息后将消息从队列中删除

  如果rabbitmq向消费者改善消息时,消费者服务器挂了,消息也不会超时,即使一个消息需要非常长的时间处理,也不会导致消息超时,永远不会从rabbitmq中删除,

忘记通过basicAck返回确认信息是个严重的错误

rabbitmq不允许重新定义一个已有的队列信息

QueueingConsumer

队列消费者,用于监听队列中的消息。调用nextDelivery方法时,内部实现就是调用队列的take方法。该方法的作用:获取并移除此队列的头部,在元素变得可用之前一直等待(如果有必要)。说白了就是如果没有消息,就处于阻塞状态。

RabbitMQ 笔记-Exchanges

  Procuder Publish的Message进入了Exchange。接着通过“routing keys”, RabbitMQ会找到应该把这个Message放到哪个queue里。queue也是通过这个routing keys来做的绑定。

有三种类型的Exchanges:direct, fanout,topic。 每个实现了不同的路由算法(routing algorithm)。

  Direct exchange: 如果 routing key 匹配, 那么Message就会被传递到相应的queue中。其实在queue创建时,它会自动的以queue的名字作为routing key来绑定那个exchange。

生产者:     // 声明交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 发送消息
        for (String severity : routingKeys) {
            String message = "Send the message level: " + severity;
            channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
            System.out.println(" [x] Sent ‘" + severity + "‘:‘" + message + "‘");
        }

消费者
        // 声明交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 获取匿名队列名称
        String queueName = channel.queueDeclare().getQueue();
        // 根据路由关键字进行多重绑定
        for (String severity : routingKeys) {
            channel.queueBind(queueName, EXCHANGE_NAME, severity);
            System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity);
        }

  Fanout exchange: 会向响应的queue广播。

生产者
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 分发消息
        for (int i=0; i<5; i++) {
            String message = "Hello World!" + i;
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] Sent ‘" + message + "‘");
        }

消费者
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
    

  Topic exchange: 对key进行模式匹配,比如ab*可以传递到所有ab*的queue。

生产者
            // 声明一个匹配模式的交换器
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            // 待发送的消息
            String routingKeys[] = new String[]{
                "quick.orange.rabbit",
                "lazy.orange.elephant",
                "quick.orange.fox",
                "lazy.brown.fox",
                "quick.brown.fox",
                "quick.orange.male.rabbit",
                "lazy.orange.male.rabbit"
            };

            // 发送消息
            for (String severity : routingKeys) {
                String message = "From "+severity+" routingKey‘ s message!";
                channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
                System.out.println("TopicSend [x] Sent ‘" + severity + "‘:‘" + message + "‘");
            }

消费者
        // 声明一个匹配模式的交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();

        // 路由关键字
        String routingKeys[] = new String[] {
            "*.orange.*"
        };

        // 绑定路由关键字

        for (String bindingKey : routingKeys) {
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
            System.out.println("ReceiveLogsTopic1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + bindingKey);
        }

  匿名: 直接发送到queue。

生产者
        for (int i=0; i<5; i++) {
            String message = "hello world! " + i;
            channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            System.out.println(" [x] Sent ‘" + message + "‘");
        }

消费者
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

RabbitMQ 笔记-RPC

RabbitMQ中实现RPC的机制是:

  • 客户端发送请求(消息)时,在消息的属性(MessageProperties,在AMQP协议中定义了14中properties,这些属性会随着消息一起发送)中设置两个值replyTo(一个Queue名称,用于告诉服务器处理完成后将通知我的消息发送到这个Queue中)和correlationId(此次请求的标识号,服务器处理完成后需要将此属性返还,客户端将根据这个id了解哪条请求被成功执行了或执行失败)
  • 服务器端收到消息并处理
  • 服务器端处理完消息后,将生成一条应答消息到replyTo指定的Queue,同时带上correlationId属性
  • 客户端之前已订阅replyTo指定的Queue,从中收到服务器的应答消息后,根据其中的correlationId属性分析哪条请求被执行了,根据执行结果进行后续业务处理

public class RPCClient {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    private Connection connection;
    private Channel channel;
    private String replyQueueName;
    private QueueingConsumer consumer;

    public RPCClient() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        // 设置MabbitMQ所在主机ip或者主机名
        factory.setHost("192.168.65.136");
        factory.setUsername("rabbitmq");
        factory.setPassword("123456");
        // 创建一个连接
        connection = factory.newConnection();
        // 创建一个频道
        channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

        //为每一个客户端获取一个随机的回调队列
        replyQueueName = channel.queueDeclare().getQueue();
        //为每一个客户端创建一个消费者(用于监听回调队列,获取结果)
        consumer = new QueueingConsumer(channel);
        //消费者与队列关联
        channel.basicConsume(replyQueueName, true, consumer);
    }

    /**
     * 获取斐波列其数列的值
     *
     * @param message
     * @return
     * @throws Exception
     */
    public String call(String message) throws Exception{
        String response = null;
        String corrId = java.util.UUID.randomUUID().toString();

        //设置replyTo和correlationId属性值
        BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();

        //发送消息到rpc_queue队列
        channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes());

        while (true) {
            System.out.println("OK?");
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            System.out.println("OK");
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response = new String(delivery.getBody(),"UTF-8");
                break;
            }
        }

        return response;
    }

    public static void main(String[] args) throws Exception {
        RPCClient fibonacciRpc = new RPCClient();
        String result = fibonacciRpc.call("4");
        System.out.println( "fib(4) is " + result);
    }
}

public class RPCServer {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        // 设置MabbitMQ所在主机ip或者主机名
        factory.setHost("192.168.65.136");
        factory.setUsername("rabbitmq");
        factory.setPassword("123456");
        // 创建一个连接
        Connection connection = factory.newConnection();
        // 创建一个频道
        Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

        //限制:每次最多给一个消费者发送1条消息
        channel.basicQos(1);

        //为rpc_queue队列创建消费者,用于处理请求
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

        System.out.println(" [x] Awaiting RPC requests");

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();

            //获取请求中的correlationId属性值,并将其设置到结果消息的correlationId属性中
            BasicProperties props = delivery.getProperties();
            BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
            //获取回调队列名字
            String callQueueName = props.getReplyTo();

            String message = new String(delivery.getBody(),"UTF-8");

            System.out.println(" [.] fib(" + message + ")");

            //获取结果
            String response = "" + fib(Integer.parseInt(message));
            //先发送回调结果
            channel.basicPublish("", callQueueName, replyProps,response.getBytes());
            //后手动发送消息反馈
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }

    /**
     * 计算斐波列其数列的第n项
     *
     * @param n
     * @return
     * @throws Exception
     */
    private static int fib(int n) throws Exception {
        if (n < 0)
            throw new Exception("参数错误,n必须大于等于0");
        if (n == 0)
            return 0;
        if (n == 1)
            return 1;
        return fib(n - 1) + fib(n - 2);
    }
}

时间: 2024-10-12 15:20:26

RabbitMQ 学习开发笔记的相关文章

UWP学习开发笔记记录(开篇)

零零散散开发微软移动2年多了,基本上从未记录或写过任何笔记.所以打算写一些自己的心得和技术的分享,大家一起来共同探讨.虽然现在UWP的工作几乎没有了,但是我感觉大家都是在观望,再看接下来微软的动作,所以在职的小伙伴闲暇之余有时间可以来补充学习一下UWP的技术,但是全身心的投入到UWP的怀抱,暂时还不是个好时机.对于微软其他的技术,我觉得还是很不错的,最近本人的前同事及好友有很多都放弃.NET去做其他语言开发了,感觉进入到了全民抵制微软技术的时代,也许这就是微软所谓的转型带来的阵痛吧,不过本人目前

Android 学习开发笔记《Android认识》

1.          Android:2007年11月5日第一版,2009年5月豪华版,2010年HTC手机 2.          Android框架主要:应用程序.应用程序框架.函数库.运行时.Linux内核 3.          应用程序:通讯录.短信.日历.天气等 4.          应用程序框架:Android 提供的API 5.          函数库:C/C++ 库,一般不能直接使用,需要封装成组件,程序员调用API 6.          运行时:核心库.Dalvik系

Android 学习开发笔记《Service 与 Thread 的区别 》

很多时候,你可能会问,为什么要用 Service,而不用 Thread 呢,因为用 Thread 是很方便的,比起 Service 也方便多了,下面我详细的来解释一下. 1). Thread:Thread 是程序执行的最小单元,它是分配CPU的基本单位.可以用 Thread 来执行一些异步的操作. 2). Service:Service 是android的一种机制,当它运行的时候如果是Local Service,那么对应的 Service 是运行在主进程的 main 线程上的.如:onCreat

Android开发笔记(一百一十六)网络学习资源

知名网站 本系列的开发笔记,对Android开发来说只是沧海一瓢,还有更多的技术等待我们去汲取.下面列出几个常用的开发网站,供初学者上路: 首先当然是国内首屈一指的技术网站csdn啦,csdn提供了众多频道,包括博客.论坛.下载.问答等等,其中博客专栏提供了最新的技术文章,值得推荐.csdn博客专栏的地址是 http://blog.csdn.net/column.html 下面是csdn博客专栏的网页截图: 其次是国外有名的开源网站GitHub,这里有众多的开源项目源码,是开发者分享代码的乐园.

RabbitMQ学习笔记五:RabbitMQ之优先级消息队列

RabbitMQ优先级队列注意点: 1.只有当消费者不足,不能及时进行消费的情况下,优先级队列才会生效 2.RabbitMQ3.5以后才支持优先级队列 代码在博客:RabbitMQ学习笔记三:Java实现RabbitMQ之与Spring集成 最后面有下载地址,只是做了少许改变,改变的代码如下: 消费者 spring-config.xml(还需要增加一个QueueListener监听器,代码就不复制到这里了,可以参考项目中的其他监听器) <!-- =========================

PHP微信公众开发笔记(二)

PHP微信公众开发笔记系列 日期:2014.9.1 需求分析: 在成功的开启了微信开发者服务之后,我们便需要开始做响应用户操作的功能了.其实做微信公众平台,当初的目的我就是想做成一个服务平台,因为刚大学毕业,同学各奔东西.有的时候可能我们会到一个陌生的城市去,那时可能就会想啊,这个城市有班上的同学在么?这时掏出手机,点开服务号输入这个城市名,就会返回一串信息,假如有同学在这个城市,就会列出联系信息之类的.当然也许会说,既然是熟悉的同学,那么肯定会知道联系方式之类的,还需要这样多此一举么.而我的考

学习hibernate笔记

以前学习java的时候,一开始就学习了hibernate,那时候总觉得ssh非常高大上,所以就急忙看了下相关视频.不过因为实际需要不高,所以后来一直没有使用上hibernate组件.现在一年过去了,也疯狂学习了java一段时间了,做过几个不大的项目,但是总算对java有些了解.现在参加了工作,公司使用的就是ssh,所以这两天又重新开始捣鼓hibernate.这次学习直接使用editplus,直接开发.看了官网的demo,发现英语也没有想象中那么困难.哈哈,把自己的学习记录下来吧.这里主要记录三个

IOS开发笔记-百度地图(第三方库)

最近做了百度地图,在导入SDK后遇到了一些问题 编译错误: linker command failed with exit code 1 (use -v to see invocation) 想了很多办法,查了很多资料,最后终于解决. 可能原因: 1. 有重复的.m文件,或者未导入 解决办法:有重复的删除即可 在工作左边导航栏Target-->Build Phases-->compile Sources中,第三库库的所有.m文件都添加到里面 2.Valid Architectures 的值 在

ArcGIS AddIn开发笔记(一)

学习AddIn开发,遇到了些稀奇古怪的问题,网上的资料少之又少. (1)AddIn开发,主要是通过ArcMap静态变量,与主程序中的数据等进行交互 (2)failed to register Add In .esriAddIn异常,此类异常为Visual Studio中项目名称或者类名称中出现了中文名 (3)AddIn个人觉得很不稳定,第一次写完一个Tool,在其MouseDown事件中写完了相应的代码,设置断点,结果每次都不断,也不执行,无奈的很.下载了几个网上的程序,设置了断点就断了,我重新