pulsar: 批量接收消息

接收消息时，和kafka类似，如果topic有多个分区，则只能保证分区内数据的接收有序，不能保证全局有序。

一、发送消息

package cn.edu.tju.test1;import org.apache.pulsar.client.api.*;public class BatchProducer01 {private static String SERVER_4 = "xx.xx.xx.xx";public static void main(String[] args) throws PulsarClientException {PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://" + SERVER_4 + ":6650").build();Producer<String> stringProducer = client.newProducer(Schema.STRING).topic("my-topic").create();for(int i = 0; i < 97; i ++){stringProducer.send("this is a good book." + i);}stringProducer.close();client.close();}
}

二、批量接收消息

package cn.edu.tju.test1;import org.apache.pulsar.client.api.*;public class BatchConsumer01 {private static String SERVER_4 = "xx.xx.xx.xx";public static void main(String[] args) throws Exception{BatchReceivePolicy batchReceivePolicy = new BatchReceivePolicy.Builder().maxNumMessages(5).build();PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://"  + SERVER_4 + ":6650").build();Consumer<String> consumer = client.newConsumer(Schema.STRING).batchReceivePolicy(batchReceivePolicy).subscriptionName("my-sub").topic("my-topic").subscribe();while (true){Messages<String> messages = consumer.batchReceive();System.out.println(messages.size());for(Message msg : messages){System.out.println(msg.getValue());}}}
}