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

本篇内容介绍了"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) 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的使用"的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注网站，小编将为大家输出更多高质量的实用文章！