rocket mq——批次实例

谁践踏了优雅 2022-06-08 04:52 128阅读 0赞

# 批次实例 #

#### 为什么批量？ ####

批量发送消息提高了传递小消息的性能。

#### 使用限制 ####

相同批次的消息应该具有：相同的主题，相同的waitStoreMsgOK和没有计划支持。

另外一批邮件的总大小不能超过1MiB。

#### 如何使用批次 ####

如果您一次发送的消息不超过1MiB，则容易使用批处理：

String topic = "BatchTest";
    List<Message> messages = new ArrayList<>();
    messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
    messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
    messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
    try { 
        producer.send(messages);
    } catch (Exception e) { 
        e.printStackTrace();
        //handle the error
    }

#### 分成列表 ####

当您发送大批量时，复杂性只会增加，您可能无法确定它是否超过了大小限制（1MiB）。

在这个时候，你最好分割列表：

public class ListSplitter implements Iterator<List<Message>> { 
        private final int SIZE_LIMIT = 1000 * 1000;
        private final List<Message> messages;
        private int currIndex;
        public ListSplitter(List<Message> messages) { 
                this.messages = messages;
        }
        @Override public boolean hasNext() { 
            return currIndex < messages.size();
        }
        @Override public List<Message> next() { 
            int nextIndex = currIndex;
            int totalSize = 0;
            for (; nextIndex < messages.size(); nextIndex++) { 
                Message message = messages.get(nextIndex);
                int tmpSize = message.getTopic().length() + message.getBody().length;
                Map<String, String> properties = message.getProperties();
                for (Map.Entry<String, String> entry : properties.entrySet()) { 
                    tmpSize += entry.getKey().length() + entry.getValue().length();
                }
                tmpSize = tmpSize + 20; //for log overhead
                if (tmpSize > SIZE_LIMIT) { 
                    //it is unexpected that single message exceeds the SIZE_LIMIT
                    //here just let it go, otherwise it will block the splitting process
                    if (nextIndex - currIndex == 0) { 
                       //if the next sublist has no element, add this one and then break, otherwise just break
                       nextIndex++;  
                    }
                    break;
                }
                if (tmpSize + totalSize > SIZE_LIMIT) { 
                    break;
                } else { 
                    totalSize += tmpSize;
                }
        
            }
            List<Message> subList = messages.subList(currIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }
    }
    //then you could split the large list into small ones:
    ListSplitter splitter = new ListSplitter(messages);
    while (splitter.hasNext()) { 
       try { 
           List<Message>  listItem = splitter.next();
           producer.send(listItem);
       } catch (Exception e) { 
           e.printStackTrace();
           //handle the error
       }
    }