本篇内容介绍了“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的使用”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站,小编将为大家输出更多高质量的实用文章!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。