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

• 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