K60至尊版剛預(yù)熱 一加Ace2 Pro正面硬剛
Redmi這邊剛?cè)缁鹑巛钡男麄髁薑60 Ultra的各種技術(shù)和硬件配置，作為競(jìng)品的一加也坐不住了。一加中國(guó)區(qū)總裁李杰發(fā)布了兩條微博，表示在自家的一加Ace2上早就已經(jīng)采用了和PixelWo
public class RabbitMQTransactionDemo { private static final String QUEUE_NAME = "transaction_queue"; public static void main(String[] args) { try { // 創(chuàng)建連接工廠 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); // 創(chuàng)建連接 Connection connection = factory.newConnection(); // 創(chuàng)建信道 Channel channel = connection.createChannel(); // 聲明隊(duì)列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); try { // 開(kāi)啟事務(wù) channel.txSelect(); // 發(fā)送消息 String message = "Hello, RabbitMQ!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); // 提交事務(wù) channel.txCommit(); } catch (Exception e) { // 事務(wù)回滾 channel.txRollback(); e.printStackTrace(); } // 關(guān)閉信道和連接 channel.close(); connection.close(); } catch (Exception e) { e.printStackTrace(); } }}public class RocketMQTransactionDemo { public static void main(String[] args) throws Exception { // 創(chuàng)建事務(wù)消息生產(chǎn)者 TransactionMQProducer producer = new TransactionMQProducer("group_name"); producer.setNamesrvAddr("localhost:9876"); // 設(shè)置事務(wù)監(jiān)聽(tīng)器 producer.setTransactionListener(new TransactionListener() { @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { // 執(zhí)行本地事務(wù)邏輯，根據(jù)業(yè)務(wù)邏輯結(jié)果返回相應(yīng)的狀態(tài) // 返回 LocalTransactionState.COMMIT_MESSAGE 表示事務(wù)提交 // 返回 LocalTransactionState.ROLLBACK_MESSAGE 表示事務(wù)回滾 // 返回 LocalTransactionState.UNKNOW 表示事務(wù)狀態(tài)未知 } @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { // 根據(jù)消息的狀態(tài)，來(lái)判斷本地事務(wù)的最終狀態(tài) // 返回 LocalTransactionState.COMMIT_MESSAGE 表示事務(wù)提交 // 返回 LocalTransactionState.ROLLBACK_MESSAGE 表示事務(wù)回滾 // 返回 LocalTransactionState.UNKNOW 表示事務(wù)狀態(tài)未知 } }); // 啟動(dòng)事務(wù)消息生產(chǎn)者 producer.start(); // 構(gòu)造消息 Message msg = new Message("topic_name", "tag_name", "Hello, RocketMQ!".getBytes()); // 發(fā)送事務(wù)消息 TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null); System.out.println("Send Result: " + sendResult); // 關(guān)閉事務(wù)消息生產(chǎn)者 producer.shutdown(); }}public class KafkaTransactionDemo { public static void main(String[] args) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional_id"); Producer<String, String> producer = new KafkaProducer<>(props); // 初始化事務(wù) producer.initTransactions(); try { // 開(kāi)啟事務(wù) producer.beginTransaction(); // 發(fā)送消息 ProducerRecord<String, String> record = new ProducerRecord<>("topic_name", "Hello, Kafka!"); producer.send(record); // 提交事務(wù) producer.commitTransaction(); } catch (ProducerFencedException e) { // 處理異常情況 producer.close(); } finally { producer.close(); } }}RabbitMQ使用ACK（消息確認(rèn)）機(jī)制來(lái)確保消息的可靠傳遞。消費(fèi)者收到消息后，需要向RabbitMQ發(fā)送ACK來(lái)確認(rèn)消息的處理狀態(tài)。只有在收到ACK后，RabbitMQ才會(huì)將消息標(biāo)記為已成功傳遞，否則會(huì)將消息重新投遞給其他消費(fèi)者或者保留在隊(duì)列中。
以下是RabbitMQ ACK的Java示例：
public class RabbitMQAckDemo { public static void main(String[] args) throws Exception { // 創(chuàng)建連接工廠 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); // 創(chuàng)建連接 Connection connection = factory.newConnection(); // 創(chuàng)建信道 Channel channel = connection.createChannel(); // 聲明隊(duì)列 String queueName = "queue_name"; channel.queueDeclare(queueName, false, false, false, null); // 創(chuàng)建消費(fèi)者 String consumerTag = "consumer_tag"; boolean autoAck = false; // 關(guān)閉自動(dòng)ACK // 消費(fèi)消息 channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // 消費(fèi)消息 String message = new String(body, "UTF-8"); System.out.println("Received message: " + message); try { // 模擬處理消息的業(yè)務(wù)邏輯 processMessage(message); // 手動(dòng)發(fā)送ACK確認(rèn)消息 long deliveryTag = envelope.getDeliveryTag(); channel.basicAck(deliveryTag, false); } catch (Exception e) { // 處理消息異常，可以選擇重試或者記錄日志等操作 System.out.println("Failed to process message: " + message); e.printStackTrace(); // 手動(dòng)發(fā)送NACK拒絕消息，并可選是否重新投遞 long deliveryTag = envelope.getDeliveryTag(); boolean requeue = true; // 重新投遞消息 channel.basicNack(deliveryTag, false, requeue); } } }); } private static void processMessage(String message) { // 模擬處理消息的業(yè)務(wù)邏輯 }}RocketMQ的ACK機(jī)制由消費(fèi)者控制，消費(fèi)者從消息隊(duì)列中消費(fèi)消息后，可以手動(dòng)發(fā)送ACK確認(rèn)消息的處理狀態(tài)。只有在收到ACK后，RocketMQ才會(huì)將消息標(biāo)記為已成功消費(fèi)，否則會(huì)將消息重新投遞給其他消費(fèi)者。
public class RocketMQAckDemo { public static void main(String[] args) throws Exception { // 創(chuàng)建消費(fèi)者 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_name"); consumer.setNamesrvAddr("localhost:9876"); // 訂閱消息 consumer.subscribe("topic_name", "*"); // 注冊(cè)消息監(jiān)聽(tīng)器 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { for (MessageExt message : msgs) { try { // 消費(fèi)消息 String msgBody = new String(message.getBody(), "UTF-8"); System.out.println("Received message: " + msgBody); // 模擬處理消息的業(yè)務(wù)邏輯 processMessage(msgBody); // 手動(dòng)發(fā)送ACK確認(rèn)消息 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } catch (Exception e) { // 處理消息異常，可以選擇重試或者記錄日志等操作 System.out.println("Failed to process message: " + new String(message.getBody())); e.printStackTrace(); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }); // 啟動(dòng)消費(fèi)者 consumer.start(); } private static void processMessage(String message) { // 模擬處理消息的業(yè)務(wù)邏輯 }}Kafka的ACK機(jī)制用于控制生產(chǎn)者在發(fā)送消息后，需要等待多少個(gè)副本確認(rèn)才視為消息發(fā)送成功。這個(gè)機(jī)制可以通過(guò)設(shè)置acks參數(shù)來(lái)進(jìn)行配置。
下面是一個(gè)使用Java編寫(xiě)的Kafka生產(chǎn)者示例代碼：
public class KafkaProducerDemo { public static void main(String[] args) { // 配置Kafka生產(chǎn)者的參數(shù) Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); // Kafka集群的地址和端口 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 鍵的序列化器 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 值的序列化器 props.put("acks", "all"); // 設(shè)置ACK機(jī)制為所有副本都確認(rèn) // 創(chuàng)建生產(chǎn)者實(shí)例 KafkaProducer<String, String> producer = new KafkaProducer<>(props); // 構(gòu)造消息 String topic = "my_topic"; String key = "my_key"; String value = "Hello, Kafka!"; // 創(chuàng)建消息記錄 ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value); // 發(fā)送消息 producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception != null) { System.err.println("發(fā)送消息出現(xiàn)異常：" + exception.getMessage()); } else { System.out.println("消息發(fā)送成功！位于分區(qū) " + metadata.partition() + "，偏移量 " + metadata.offset()); } } }); // 關(guān)閉生產(chǎn)者 producer.close(); }}本文鏈接：http://m.www897cc.com/showinfo-26-10561-0.htmlMQ黃金三劍客 Rabbit Rocket Kafka深入解密常見(jiàn)問(wèn)題及功能對(duì)比指南
聲明：本網(wǎng)頁(yè)內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問(wèn)題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請(qǐng)聯(lián)系：2376512515@qq.com 津ICP備18001702號(hào)
津公網(wǎng)安備 12010102000574號(hào)