RabbitMq Consumer和Producer端使用
程序员文章站
2022-05-23 07:55:46
...
- RabbitMq简介
- RabbitMq exchange四种方式
- RabbitMq Consumer和Producer端使用
- RabbitMq确认机制
- RabbitMq Cluster 和 High Availability
一、Producer端
ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); String exchangeName = "hello_exchange"; String routingKey = "hello_routingKey"; channel.exchangeDeclare(exchangeName, "direct", true); //定义exchange channel.queueBind(QUEUE_NAME, exchangeName, routingKey); //通过routingKey把queue和exchange绑定 byte[] messageBodyBytes = "Hello, world!".getBytes(); channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes); //发布消息
流控
RabbitMQ 服务端有自我保护机制——流控,流控分两种:
1. 全局流控
当内存/磁盘使用量达到流控阀值,会触发消息发送全局流控。发生流控后该实例所有队列都无法投递消息只能消费消息;
2. 队列流控
队列由于生产速率远远超过消费速率,同时造成该队列消息堆积严重会发生队列级别流控。发生流控后当前队列将无法投递消息,只能消费消息。
当堆积的消息数量下降后,队列流控状态会消失,消息投递将恢复正常。
所以用户当收到流控报警后要及时查看实例消费者消费能力,通过增加消费者,或者清理可丢弃消息数据等措施尽快减少消息堆积,同时应用需要对因为流控消息投递失败的情况进行处理,消息先放在其他介质或者等待重试等机制避免消息因为投递失败后消息丢失。
二、Consumer端
- Push模式:客户端被动收数据,Broker会把message轮流发给不同的Consumer,客户端接受消息不可控。即使没有ack,也会继续轮流发送。客户端默认1000大小的队列VariableLinkedBlockingQueue收取消息,然后Runtime.getRuntime().availableProcessors() * 2大小的固定大小ThreadPoolExecutor消费消息。
- Pull模式:客户端主动收数据,客户端接受消息可控。无论有没有ack,还是会继续收到message。
- 相同点:都是长链接
Push模式
即Consumer订阅queue,当有message发布到queue时,会马上push到Consumer。
ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); final Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [y] Received '" + message + "'"); this.getChannel().basicAck(envelope.getDeliveryTag(), false); //手动ack } }; channel.basicConsume(QUEUE_NAME, false, consumer); //第二个参数设置是否自动ack
Pull模式
即Consumer主动去queue拉取message。
ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); while (true) { GetResponse response = channel.basicGet(QUEUE_NAME, false); //第二个参数设置是否自动ack if (response == null) { // No message retrieved. } else { byte[] body = response.getBody(); String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + message + "'"); long deliveryTag = response.getEnvelope().getDeliveryTag(); channel.basicAck(deliveryTag, false); // acknowledge receipt of the message } }
上一篇: 我想让姐夫陪我去一趟医院
下一篇: 百度云会员破解,全速下载无限制
推荐阅读
-
浅谈pc和移动端的响应式的使用
-
Linux下安装Memcached服务器和客户端与PHP使用示例
-
使用gd库实现php服务端图片裁剪和生成缩略图功能分享
-
【翻译】使用WebApi和Asp.Net Core Identity 认证 Blazor WebAssembly(Blazor客户端应用)
-
从淘宝和网易的font-size思考移动端怎样使用rem?
-
使用vue实现简单键盘的示例(支持移动端和pc端)
-
移动端使用CSS或JS判断横屏和竖屏的讲解
-
.net下redis和rabbitmq简单使用demo
-
node.js使用http模块创建服务器和客户端完整示例
-
在移动端使用vue-router和keep-alive的方法示例