其他分享
首页 > 其他分享> > 生产者MessageQueueSelector实战

生产者MessageQueueSelector实战

作者:互联网

 top下面默认有四个Queue, queque的数量不能大约配置,否者会报错

 

 

假设一个top下面有三个类目,分别是手机,衣服,食品,他们发送消息都是随机发送到一个queue里面,如果有一天,衣服的消息突然增多了,堵塞队列了,其他两个类目也会受到影响,造成消息发送失败,这个的话就可以指定类目发送到哪个queue,手机指定发送到队列0 ,衣服发送队列1,这样即使衣服的消息增多了也不会影响其他队列,但这样也会失去负载均衡

 

 

 

代码案例:

同步发送

@Override
public MtopResult api(MtopInnerRequest request) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
List<Produce> list = Produce.produceList();

//MessageQueueSelector 选择
if (! CollectionUtils.isEmpty(list)){
for (Produce produce : list) {
Message message = new Message("box","orderMessage",produce.getId(),JSON.toJSONString(produce).getBytes());

SendResult send = payProduct.getProducer().send(message, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
Integer queusNum = Integer.valueOf(o.toString());
return list.get(queusNum);
}
}, 0);
System.out.printf("发送结果=%s, msg=%s ", send.getSendStatus(), send.toString());
}
}

 

 

top 默认是4个queue

 

 

queue不能大于配置,

 

 

 报错

 

 

异步发送

 

//MessageQueueSelector 选择
if (!CollectionUtils.isEmpty(list)) {

for (Produce produce : list) {
Message message = new Message("box", "orderMessage", produce.getId(), JSON.toJSONString(produce).getBytes());
payProduct.getProducer().send(message, (list1, message1, o) ->
{
Integer integer = Integer.valueOf(o.toString());
return list1.get(integer);
}
, 4, new SendCallback() {

@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());

}

@Override
public void onException(Throwable throwable) {
System.out.println(throwable);
throwable.printStackTrace();
}
});
}


 

 




}

 

标签:实战,生产者,message,list,send,发送,produce,Message,MessageQueueSelector
来源: https://www.cnblogs.com/HuangXingLei/p/12620242.html