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