環(huán)境: springboot2.3.9RELEASE RocketMQ4.8.0

依存関係

<dependency>???<groupid>org.springframework.boot</groupid>?????<artifactid>spring-boot-starter-web</artifactid>?</dependency>?<dependency>?????<groupid>org.apache.rocketmq</groupid>?????<artifactid>rocketmq-spring-boot-starter</artifactid>?????<version>2.2.0</version>?</dependency>

設(shè)定ファイル

server:???port:?8080?---?rocketmq:???nameServer:?localhost:9876???producer:?????group:?demo-mq

通常のメッセージ

送信

@Resource?private?RocketMQTemplate?rocketMQTemplate?;??????public?void?send(String?message)?{???rocketMQTemplate.convertAndSend("test-topic:tag2",?MessageBuilder.withPayload(message).build());?}

Accept

@RocketMQMessageListener(topic?=?"test-topic",?consumerGroup?=?"consumer01-group",?selectorExpression?=?"tag1?||?tag2")?@Component?public?class?ConsumerListener?implements?RocketMQListener<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println("接收到消息："?+?message)?;?????}??}</string>

連続メッセージ

Send

@Resource?private?RocketMQTemplate?rocketMQTemplate?;??public?void?sendOrder(String?topic,?String?message,?String?tags,?int?id)?{?????rocketMQTemplate.asyncSendOrderly(topic?+?":"?+?tags,?MessageBuilder.withPayload(message).build(),??????????????"order-"?+?id,?new?SendCallback()?{?????????????????@Override?????????????????public?void?onSuccess(SendResult?sendResult)?{?????????????????????System.err.println("msg-id:?"?+?sendResult.getMsgId()?+?":?"?+?message?+"\tqueueId:?"?+?sendResult.getMessageQueue().getQueueId())?;?????????????????}?????????????????@Override?????????????????public?void?onException(Throwable?e)?{?????????????????????e.printStackTrace()?;?????????????????}?????????????});?}

ハッシュキー

@RocketMQMessageListener(topic?=?"order-topic",?consumerGroup?=?"consumer02-group",??????selectorExpression?=?"tag3?||?tag4",?consumeMode?=?ConsumeMode.ORDERLY)?@Component?public?class?ConsumerOrderListener?implements?RocketMQListener<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println(Thread.currentThread().getName()?+?"?接收到Order消息："?+?message)?;?????}??}</string>

consumeMode = ConsumeMode.ORDERLYに基づいて別のキューに送信されるメッセージを示します。メッセージ モードはシーケンシャル モードで、1 つのキューと 1 つのスレッドです。

Result

consumeMode = ConsumeMode.CONCURRENTLY の場(chǎng)合、実行結(jié)果は次のようになります。

# #クラスター/ブロードキャスト メッセージ モード

送信者

@Resource?private?RocketMQTemplate?rocketMQTemplate?;??????public?void?send(String?topic,?String?message,?String?tags)?{?????rocketMQTemplate.send(topic?+?":"?+?tags,?MessageBuilder.withPayload(message).build())?;?}

クラスター メッセージ モード

コンシューマー

@RocketMQMessageListener(topic?=?"broad-topic",?consumerGroup?=?"consumer03-group",??????selectorExpression?=?"tag6?||?tag7",?messageModel?=?MessageModel.CLUSTERING)?@Component?public?class?ConsumerBroadListener?implements?RocketMQListener<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println("ConsumerBroadListener1接收到消息："?+?message)?;?????}??}</string>

messageModel = MessageModel.CLUSTERING

Test

ポート 8080 と 8081 で 2 つのサービスを開始します

8080 サービス

8081 サービス

クラスター メッセージ モードでは、各サービスは負(fù)荷分散を?qū)g現(xiàn)するためにメッセージの一部を個(gè)別に受信します。

ブロードキャスト メッセージ モード

コンシューマー エンド

@RocketMQMessageListener(topic?=?"broad-topic",?consumerGroup?=?"consumer03-group",??????selectorExpression?=?"tag6?||?tag7",?messageModel?=?MessageModel.BROADCASTING)?@Component?public?class?ConsumerBroadListener?implements?RocketMQListener<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println("ConsumerBroadListener1接收到消息："?+?message)?;?????}??}</string>

messageModel = MessageModel。ブロードキャスティング

テスト

ポート 8080 と 8081 で 2 つのサービスを開始します

8080 サービス

8081 サービス

#クラスター メッセージ モードでは、各サービスは同じメッセージを受信します。

トランザクション メッセージ

RocketMQ トランザクションの 3 つのステータス

TransactionStatus.CommitTransaction: トランザクション メッセージを送信し、コンシューマーはこのメッセージを消費(fèi)できます

TransactionStatus.RollbackTransaction: ローリングトランザクションを戻すということは、メッセージが削除され、使用できなくなることを意味します。

TransactionStatus.Unknown: 中間ステータス。ステータスを判斷するためにメッセージ キューをチェックする必要があることを表します。

RocketMQ のトランザクション メッセージの実裝は、主に 2 つの段階に分かれています: 通常のトランザクションの送信と送信、およびトランザクション情報(bào)の補(bǔ)償プロセスです。全體のプロセスは次のとおりです:

通常のトランザクションの送信と送信段階

1. プロデューサは MQServer にハーフ メッセージを送信します (ハーフ メッセージとは、コンシューマが一時(shí)的に消費(fèi)できないメッセージを指します)

2. サーバーはメッセージの書き込み結(jié)果に応答し、ハーフ メッセージが送信されます正常に完了しました

3. ローカルトランザクションの実行を開始します

4. ローカルトランザクションの実行狀況に応じてCommitまたはRollbackを?qū)g行します

トランザクション情報(bào)の補(bǔ)償処理

1. MQServer が長(zhǎng)期間受信しない場(chǎng)合、ローカル トランザクションの実行ステータスにより、プロデューサーへの確認(rèn)レビュー操作リクエストが開始されます。

2. プロデューサが確認(rèn)レビュー リクエストを受信した後、ローカル トランザクションの実行ステータスを確認(rèn)します。

3. 「結(jié)果を確認(rèn)した後、コミットまたはロールバック操作を?qū)g行します?！工摔瑜毪?、

#補(bǔ)償フェーズは主にタイムアウトまたはロールバックの問題を解決するために使用されます。プロデューサがコミットまたはロールバック操作を送信すると失敗します。

Sender

@Resource?private?RocketMQTemplate?rocketMQTemplate?;??????public?void?sendTx(String?topic,?Long?id,?String?tags)?{?????rocketMQTemplate.sendMessageInTransaction(topic?+?":"?+?tags,?MessageBuilder.withPayload(?????????????new?Users(id,?UUID.randomUUID().toString().replaceAll("-",?""))).?????????????setHeader("BID",?UUID.randomUUID().toString().replaceAll("-",?"")).build(),??????????????UUID.randomUUID().toString().replaceAll("-",?""))?;?}

プロデューサーに対応するリスナー

@RocketMQTransactionListener?public?class?ProducerTxListener?implements?RocketMQLocalTransactionListener?{??????????@Resource?????private?BusinessService?bs?;??????@Override?????public?RocketMQLocalTransactionState?executeLocalTransaction(Message?msg,?Object?arg)?{?????????//?這里執(zhí)行本地的事務(wù)操作，比如保存數(shù)據(jù)。?????????try?{?????????????//?創(chuàng)建一個(gè)日志記錄表，將這唯一的ID存入數(shù)據(jù)庫中，在下面的check方法中可以根據(jù)這個(gè)id查詢是否有數(shù)據(jù)?????????????String?id?=?(String)?msg.getHeaders().get("BID")?;?????????????Users?users?=?new?JsonMapper().readValue((byte[])msg.getPayload(),?Users.class)?;?????????????System.out.println("消息內(nèi)容："?+?users?+?"\t參與數(shù)據(jù)："?+?arg?+?"\t本次事務(wù)的唯一編號(hào)："?+?id)?;?????????????bs.save(users,?new?UsersLog(users.getId(),?id))?;?????????}?catch?(Exception?e)?{?????????????e.printStackTrace()?;?????????????return?RocketMQLocalTransactionState.ROLLBACK?;?????????}?????????return?RocketMQLocalTransactionState.COMMIT?;?????}??????@Override?????public?RocketMQLocalTransactionState?checkLocalTransaction(Message?msg)?{?????????//?這里檢查本地事務(wù)是否執(zhí)行成功?????????String?id?=?(String)?msg.getHeaders().get("BID")?;?????????System.out.println("執(zhí)行查詢ID為："?+?id?+?"?的數(shù)據(jù)是否存在")?;?????????UsersLog?usersLog?=?bs.queryUsersLog(id)?;?????????if?(usersLog?==?null)?{?????????????return?RocketMQLocalTransactionState.ROLLBACK?;?????????}?????????return?RocketMQLocalTransactionState.COMMIT?;?????}??}

Consumer

@RocketMQMessageListener(topic?=?"tx-topic",?consumerGroup?=?"consumer05-group",?selectorExpression?=?"tag10")?@Component?public?class?ConsumerTxListener?implements?RocketMQListener<users>?{??????@Override?????public?void?onMessage(Users?users)?{?????????System.out.println("TX接收到消息："?+?users)?;?????}??}</users>

Service

@Transactional?public?boolean?save(Users?users,?UsersLog?usersLog)?{?????usersRepository.save(users)?;?????usersLogRepository.save(usersLog)?;?????if?(users.getId()?==?1)?{?????????throw?new?RuntimeException("數(shù)據(jù)錯(cuò)誤")?;?????}?????return?true?;?}??????public?UsersLog?queryUsersLog(String?bid)?{?????return?usersLogRepository.findByBid(bid)?;?}

Controller

@GetMapping("/tx/{id}")?public?Object?sendTx(@PathVariable("id")Long?id)?{?????ps.sendTx("tx-topic",?id,?"tag10")?;?????return?"send?transaction?success"?;?}

Test

インターフェイスを呼び出した後のコンソール出力:

印刷ログから、コンシューマはすべての処理が完了した後でのみメッセージを受信することがわかります。メッセージが保存されました。

データを削除してから ID を 1 としてテストすると、エラーが発生します。

#データベースにデータがありません。 。 。

非常に複雑ではありませんか? 2 段階で処理できます。

以上がSpringBoot が RocketMQ トランザクション、ブロードキャスト、連続メッセージを統(tǒng)合する方法の詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語 Web サイトの他の関連記事を參照してください。