温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

rabbitmq的基本概念以及amqp-client的使用

发布时间:2021-07-02 17:04:12 来源:亿速云 阅读:364 作者:chen 栏目:大数据

本篇内容介绍了“rabbitmq的基本概念以及amqp-client的使用”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!

package com.rabbitmq.demo;

import com.rabbitmq.client.*;

public class Produce {

    public static void main(String args[]) throws Exception{
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setPassword("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setHost("192.168.1.141");
        connectionFactory.setPort(5672);
        Connection connection=connectionFactory.newConnection();
        Channel channel= connection.createChannel();
        String exchangeName="test";
        String queueName="test";
        String routingkey="routingkey";
        String bindingkey=routingkey;
        //设置交换机,直连,持久化,非自动删除,参数
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,null);
        //channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);
        /**
         * BuiltinExchangeType.FANOUT  不需要绑定 bindingkey与routingkey exchange直接将消息投递到queue
         * BuiltinExchangeType.DIRECT  需要绑定  且bindingkey与routingkey一致时 exchange投递消息到queue
         * BuiltinExchangeType.TOPIC   需要绑定  模糊匹配,规则如下
             * RoutingKey 为一个点号"."分隔的字符串(被点号"。"分隔开的每段独立的字符,如com.rabbit.demo,com.rabbit.test)
             * BindingKey RoutingKey 样也是点号"."分隔的字符串;
             * BindingKey 中可以存在两种特殊字符串"*"和"#",用于做模糊匹配,其中"#"用于匹配多个(零个到多个),* 匹配任意一个单词。
         *    如果binding_key 是 “#” - 它会接收所有的Message,不管routing_key是什么,就像是fanout exchange。
         *    如果 “*” and “#” 没有被使用,那么topic exchange就变成了direct exchange。
         * BuiltinExchangeType.HEADERS 不需要绑定
         *   消费者arguments指定“x-match”,这个键的Value可以是any或者all,这代表消息携带的Hash是需要全部匹配(all),还是仅匹配一个键(any)就可以了
         */
        /**
         * 持久化:数据会保存到磁盘,重启rabbitmq数据依旧存在
         *
         * 排他:该队列仅对首次声明它的连接可见,并在连接断开时自动删除。
         * 需要注意三点:排他队列是基于连接( Connection) 可见的,同一个连接的不同信道 (Channel)
         * 是可以同时访问同一连接创建的排他队列; "首次"是指如果一个连接己经声明了
         * 排他队列,其他连接是不允许建立同名的排他队列的,这个与普通队列不同:即使该队
         * 列是持久化的,一旦连接关闭或者客户端退出,该排他队列都会被自动删除,这种队列
         * 适用于一个客户端同时发送和读取消息的应用场景。
         *
         *
         * 自动删除:设置是否自动删除。为 true 则设置队列为自动删除。自动删除的前提是:
         * 至少有一个消费者连接到这个队列,之后所有与这个队列连接的消费者都断开时,才会
         * 自动删除。不能把这个参数错误地理解为: "当连接到此队列的所有客户端断开时,这
         * 个队列自动删除",因为生产者客户端创建这个队列,或者没有消费者客户端与这个队
         * 列连接时,都不会自动删除这个队列。
         *
         * 根据业务数据,最好提前建好exchange,queue,及绑定关系,生产端,消费端可以避免很多错误,如exchang创建失败,
         * 绑定关系不确定导致消息投递失败
         *
         * rabbitmq  clinet 属性集
         * props 消息的基本属性集,其包含 14 个属性成员,分别有 contentType
         * content ncoding headers Map<String Object>) deliveryMode priority
         * correlationld replyTo expiration messageld timestamp type userld
         * appld cluster 。
         *
         *   mandatory 参数设为 true 时,交换器无法根据自身的类型和路由键的队列,
         *   那么 RabbitM 会调用 Basic.Return 命令将消息返回给生产者 数设置为 false 时,出现上述情形,
         *   则消息直接被丢弃 那么生产者如何获取到没有被正确路由到合适队列的消息呢?这时channel addReturnListener 来添加 ReturnListener 监昕器实现。          *   通过rabbitmq备份队列,当消息发送不到主队列,将自动发送到备份队列,设置了备份队列,mandatory参数将无效 */


        //设置队列,持久化,非排他,非自动删除
        channel.queueDeclare(queueName,true,false,false,null);
        //channel.queueBind(queueName,exchangeName, bindingkey);
        for (int i = 0; i <10 ; i++) {
            channel.basicPublish(exchangeName,routingkey,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,"hi,Rabbitmq".getBytes() );
        }
        channel.close();
        connection.close();
    }
}
package com.rabbitmq.demo;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Consumer {

    public static void main(java.lang.String args[]) throws Exception{
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setPassword("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setHost("192.168.1.141");
        connectionFactory.setPort(5672);
        Connection connection=connectionFactory.newConnection();
        Channel channel= connection.createChannel();
        /*while(true){
            //单个获取消息,当没有消息返回null,pull模式
            GetResponse response = channel.basicGet("test001", true);
            if(response!=null){
                System.out.println(new String(response.getBody()));
            }
        }*/
        //从queue中手动确认获取消息,push模式
        /**
         * 队列名,自动签收,默认消费者
         * queue 队列的名称:
         * autoAck 设置是否自动确认。建议设成 fa se ,即不自动确认:
         *      utoAck 参数置为 false ,对于 RabbitMQ 服务端而 ,队列中的消息分成了两个部分
                 * 部分是等待投递给消费者的消息:一部分是己经投递给消费者,但是还没有收到消费者确认
                 * 信号的消息。 如果 RabbitMQ 直没有收到消费者的确认信号,并且消费此消息的消费者己经
                 * 断开连接,则 RabbitMQ 会安排该消息重新进入队列,等待投递给下 个消费者,当然也有可
                 * 能还是原来的那个消费者。
                 * RabbitMQ 会为未确认的消息设置过期时间,它判断此消息是否需要重新投递给消费者的
                 * 依据是消费该消息的消费者连接是否己经断开,这么设计的原因是 RabbitMQ 允许消费者
                 * 消费 条消息的时间可以很久很久。
         * consumerTag: 消费者标签,用来区分多个消费者:
         * noLocal 设置为 true 则表示不能将同一个 Connectio口中生产者发送的消息传送给
         * 这个 Connection 中的消费者:
         * exclusive 设置是否排他
         * arguments 设置消费者的其他参数:
         * callback 设置消费者的回调函数。用来处理 Rabb itM 推送过来的消息,比如
         * DefaultConsumer 使用时需要客户端重写 (overr e) 其中的方法。
         */
        channel.basicConsume("test",false,new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1:"+new java.lang.String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
        //创建多个消费者,默认为平均分摊round-robin
        //不支持队列层面的广播消费
        /**
         * 上面代码中显式地设置 autoAck false 然后在接收到消息之后 行显式 ack
         * (channel basicAck ), 对于消费者来说这 设置是非常 可以防止
         * 丢失。
         *
         *
         */
        channel.basicConsume("queue",false,new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2:"+new java.lang.String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
        //单个拒绝接受消息
        //channel.basicReject(111, false);
        /**
         * 多个拒绝消息
         */
       // channel.basicNack();
        /**
         * channel.basicReject 或者 channel.basicNack 中的 requeue 设直为 false ,可
         * 以启用"死信队列"的功能。死信队列可以通过检测被拒绝或者未送达的消息来追踪问题
         */
        /***
         * Basic Consume 将信道 (Channel) 直为接收模式,直到取消队列的订阅为止。在接收
         * 模式期间, RabbitMQ 会不断地推送消息给消费者,当然推送消息的个数还是会受到 Basic.Qos
         * 的限制.如果只想从队列获得单条消息而不是持续订阅,建议还是使用 Basic.Get 进行消费.但
         * 是不能将 Basic.Get 放在一个循环里来代替 Basic.Consume ,这样做会严重影响 RabbitMQ
         * 的性能.如果要实现高吞吐量,消费者理应使用 Basic.Consume 方法。
         *
         *
         */
    }
}

“rabbitmq的基本概念以及amqp-client的使用”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站,小编将为大家输出更多高质量的实用文章!

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI