欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

RabbitMQ使用场景练习:Validated User ID、Length Limit(十二 )

程序员文章站 2022-07-13 15:12:28
...
  • Validated User ID
发送消息时指定userid,只有是当前conn的用户名才可以发送出去消息(测试发现非当前conn用户名,队列没有也没有创建,可能因为是在同一个channel)。用于消费者consumer需要知道消息是从哪个用户发过来的情况

package com.demo.mq.rabbitmq.example12;
import java.io.IOException;
import java.io.Serializable;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.SerializationUtils;
import com.demo.mq.rabbitmq.MqManager;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.BasicProperties.Builder;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BlockedListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 校验User ID
 * @author sheungxin
 *
 */
public class ValidUserSend {
	private static String userId="sheungxin";
	
	/**
	 * 发送消息时校验userId
	 * @param mes
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void validUserSend(Serializable mes) throws IOException, TimeoutException{
		Connection conn=MqManager.newConnection();
		//链接堵塞监听器,除此还有ShutdownListener(非本用例的重点,略带一下)
		conn.addBlockedListener(new BlockedListener() {
			
			@Override
			public void handleUnblocked() throws IOException {
				// connection is now unblocked
			}
			
			@Override
			public void handleBlocked(String reason) throws IOException {
				// connection is now blocked
			}
		});

		Channel channel=conn.createChannel();
		String queueName=channel.queueDeclare().getQueue();
		//声明消费者用于接收消息(发在basicPublish后面也可以接收到消息,可能因为在同一个channel)
		channel.basicConsume(queueName, true, new DefaultConsumer(channel){
			@Override
		    public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException{
				System.out.println(SerializationUtils.deserialize(body));
			}
		});
		//发送消息时指定userid,只有是当前conn的用户名才可以发送出去消息(测试发现非当前conn用户名,队列没有也没有创建,可能因为是在同一个channel)
		Builder properties=new BasicProperties.Builder();
		properties.userId(userId);
		channel.basicPublish("", queueName, properties.build(), SerializationUtils.serialize(mes));
	}
	
	public static void main(String[] args) throws IOException, TimeoutException {
		validUserSend("Hello World!");
	}

}

  • Queue Length Limit
x-max-length:指队列中可存放的最大消息数量,不是队列中单条消息的最大长度
max-length-bytes:指队列中存放消息内容的最大字节长度
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-max-length", 10);
//args.put("x-max-length-bytes",10000);
channel.queueDeclare("myqueue", false, false, false, args);

可通过服务端策略设定
rabbitmqctl set_policy Ten "^one-meg$" '{"max-length-bytes":1000000}' --apply-to queues
相关标签: RabbitMQ