如何做网站栏目规划,徐州做网站费用,网站定位分析是什么,深圳大型网站建设服务接收消息时#xff0c;和kafka类似#xff0c;如果topic有多个分区#xff0c;则只能保证分区内数据的接收有序#xff0c;不能保证全局有序。
一、发送消息
package cn.edu.tju.test1;import org.apache.pulsar.client.api.*;public class BatchProducer01 {private sta…接收消息时和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();ProducerString 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();ConsumerString consumer client.newConsumer(Schema.STRING).batchReceivePolicy(batchReceivePolicy).subscriptionName(my-sub).topic(my-topic).subscribe();while (true){MessagesString messages consumer.batchReceive();System.out.println(messages.size());for(Message msg : messages){System.out.println(msg.getValue());}}}
}
