生产者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