---
schema_version: "1"
id: 830477dd4ca7808e559e448821659b50403b9e0b52100d07a0d7823425baf165
title: Apache Kafka 4.6 の例
collection: mulesoft
version: current
locale: ja-jp
portal_url: https://docs.salesforce.com/mulesoft/current/ja-jp/kafka-connector/4.6/kafka-connector-examples.md
canonical_url: https://docs.mulesoft.com/jp/kafka-connector/4.6/kafka-connector-examples
source_site: docs.mulesoft.com
content_type: documentation
content_hash: 7cd62d1050f1d20f27221b3d43f401b43761499973d778af6afc9ade48769418
description: MuleSoft Documentation Site
release: "4.6"
latest: false
---
# Apache Kafka 4.6 の例

Apache Kafka Connector の Mule フローの例を示します。

- [トピックのパブリッシュとコンシューム](#publish-and-then-consume-a-topic)

## [](#publish-and-then-consume-a-topic)トピックのパブリッシュとコンシューム

次の例は、Apache Kafka Connector の 2 つの操作 (​**Publish**​ と ​**Consume**​) を使用して、メッセージを Apache Kafka にパブリッシュしてから取得する方法を示しています。この例には次の 2 つのフローが含まれます。

- 最初のフローは、メッセージを Apache Kafka にパブリッシュするプロデューサーフローです。
- 2 番目のフローは、Apache Kafka からメッセージをコンシュームするコンシューマーフローです。

![Studio の Apache Kafka Connector のプロデューサーフロー](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-producer-studio-flow.png)

![Studio の Apache Kafka Connector のコンシューマーフロー](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-consumer-studio-flow.png)

### [](#最初のフローを設定する)最初のフローを設定する

メッセージを Apache Kafka にパブリッシュする最初のフローを設定します。最初のフローを作成するには、​**Listener**​ コンポーネント、​**Logger**​ コンポーネント、​**Publish**​ 操作、​**Set Payload**​ トランスフォーマーを設定する必要があります。

#### [](#http-リスナーを設定する)HTTP リスナーを設定する

`/pushMessage`​ パスへのコールが行われたときに Mule フローを開始するように ​**Listener**​ コンポーネントを設定します。

1. Studio で新しい Mule プロジェクトを作成します。
2. **\[Mule Palette (Mule パレット)\]**​ ビューから、​**\[HTTP\]**​ を選択して ​**\[Listener\]**​ コンポーネントをキャンバスにドラッグします。
3. プロパティウィンドウで、​**\[Connector configuration (コネクタ設定)\]**​ 項目の横にある ​**\[+\]**​ をクリックしてグローバル要素を追加します。
4. デフォルトを受け入れます。
5. プロパティウィンドウで、​**\[Path (パス)\]**​ 項目値を ​`/pushMessage`​ に設定します。

#### [](#logger-コンポーネントを追加する)Logger コンポーネントを追加する

Mule コンソールに応答を表示する ​**Logger**​ コンポーネントを追加します。

1. **\[Mule Palette (Mule パレット)\]**​ ビューから、​**\[Core (コア)\]**​ を選択して ​**\[Logger\]**​ コンポーネントを ​**\[Listener\]**​ の右側にドラッグします。
2. プロパティウィンドウで、次の項目を設定します。
    
    <table class="tableblock frame-all grid-all fit-content spread">
<colgroup>
<col>
<col>
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">項目</th>
<th class="tableblock halign-left valign-top">値</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Display Name (表示名)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">ロガーの名前。</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Message (メッセージ)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>#["Message: '"  payload.message  "' is going to be published to topic: '"  payload.topic  "'."]</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Level (レベル)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">INFO (Default) (INFO (デフォルト))</p></td>
</tr>
</tbody>
</table>
    
    次の図は、プロパティウィンドウでの ​**\[Logger\]**​ 設定の例を示しています。
    
    ![「Logger」 プロパティウィンドウ設定](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-logger-config-publish.png)

#### [](#publish-操作を追加する)\[Publish\] 操作を追加する

**\[Publish\]**​ 操作を追加して、指定した Kafka トピックにメッセージをパブリッシュします:

1. **\[Mule Palette (Mule パレット)\]**​ ビューから、​**\[Apache Kafka\]**​ を選択して ​**\[Publish\]**​ 操作を ​**\[Logger\]**​ の右側にドラッグします。
2. プロパティウィンドウで、​**\[Connector configuration (コネクタ設定)\]**​ 項目の横にある ​**\[+\]**​ をクリックしてグローバル要素を追加します。
3. 次のようにグローバル要素を設定します。
    
    1. **\[Bootstrap Server URLs (ブートストラップサーバー URL)\]**​ 項目で、​**\[Edit inline (インライン編集)\]**​ を選択し、緑色のプラス記号をクリックします。
    2. 値 ​`${config.basic.bootstrapServers}`​ を入力して ​**\[Finish (完了)\]**​ をクリックします。
4. プロパティウィンドウで、次の項目を設定します。
    
    <table class="tableblock frame-all grid-all fit-content spread">
<colgroup>
<col>
<col>
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">項目</th>
<th class="tableblock halign-left valign-top">値</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Display Name (表示名)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">コネクタ操作の名前。</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Connector configuration (コネクタ設定)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">作成したばかりのグローバル設定。</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Topic (トピック)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>\#[payload.topic]</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Key (キー)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>#[now()]</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Message (メッセージ)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>payload</code></p></td>
</tr>
</tbody>
</table>
    
    次の画像は、プロパティウィンドウでの ​**Publish**​ 設定の例を示しています。
    
    ![Publish プロパティウィンドウ設定](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-publish.png)

#### [](#set-payload-トランスフォーマーを追加する)Set Payload トランスフォーマーを追加する

応答ビルダーをプッシュする ​**Set Payload**​ トランスフォーマーを追加します。

1. **\[Mule Palette (Mule パレット)\]**​ ビューから、​**\[Core (コア)\]**​ を選択して ​**\[Set Payload\]**​ トランスフォーマーを ​**\[Publish\]**​ の右側にドラッグします。
2. プロパティウィンドウで、次の項目を設定します。
    
    <table class="tableblock frame-all grid-all fit-content spread">
<colgroup>
<col>
<col>
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">項目</th>
<th class="tableblock halign-left valign-top">値</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Display Name (表示名)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">トランスフォーマーの名前 (​<code>Push response builder</code>​ など)。</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Value (値)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>Message successfully sent to Apache Kafka topic.</code></p></td>
</tr>
</tbody>
</table>
    
    次の画像は、プロパティウィンドウでの ​**Set Payload**​ 設定の例を示しています。
    
    ![Set Payload プロパティウィンドウ設定](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-set-payload.png)

### [](#2-番目のフローを設定する)2 番目のフローを設定する

Apache Kafka からメッセージをコンシュームする 2 番目のフローを設定します。2 番目のフローを作成するには、​**Message listener**​ ソースと ​**Logger**​ コンポーネントを設定する必要があります。

#### [](#message-listener-ソースを追加する)Message Listener ソースを追加する

メッセージエンドポイントをコンシュームする ​**\[Message listener\]**​ ソースを追加します。

1. **\[Mule Palette (Mule パレット)\]**​ ビューから、​**\[Apache Kafka\]**​ を選択して ​**\[Message listener\]**​ ソースをキャンバスにドラッグします。
2. プロパティウィンドウで、​**\[Connector configuration (コネクタ設定)\]**​ 項目の横にある ​**\[+\]**​ をクリックしてグローバル要素を追加します。
3. 次のようにグローバル要素を設定します。
    
    1. **\[Bootstrap Server URLs (ブートストラップサーバー URL)\]**​ 項目で、​**\[Edit inline (インライン編集)\]**​ を選択し、緑色のプラス記号をクリックします。
    2. 値 ​`${config.basic.bootstrapServers}`​ を入力して ​**\[Finish (完了)\]**​ をクリックします。
    3. **\[Group ID (グループ ID)\]**​ 項目に ​`${consumer.groupId}`​ と入力します。
    4. **\[Topic Subscription Patterns (トピックサブスクリプションパターン)\]**​ 項目で、​**\[Edit inline (インライン編集)\]**​ を選択し、緑色のプラス記号をクリックします。
    5. 値 ​`${topic1}`​ を入力して ​**\[Finish (完了)\]**​ をクリックします。
4. プロパティウィンドウで、次の項目を設定します。
    
    <table class="tableblock frame-all grid-all fit-content spread">
<colgroup>
<col>
<col>
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">項目</th>
<th class="tableblock halign-left valign-top">値</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Display Name (表示名)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">コネクタソースの名前。</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Connector configuration (コネクタ設定)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">作成したばかりのグローバル設定。</p></td>
</tr>
</tbody>
</table>
    
    次の画像は、プロパティウィンドウでの ​**Message listener**​ 設定の例を示しています。
    
    ![Message listener プロパティウィンドウ設定](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-message-listener-config.png)

#### [](#logger-コンポーネントを追加する-2)Logger コンポーネントを追加する

Mule コンソールに応答を表示する ​**Logger**​ コンポーネントを追加します。

1. **\[Mule Palette (Mule パレット)\]**​ ビューから、​**\[Core (コア)\]**​ を選択して ​**\[Logger\]**​ コンポーネントを ​**\[Message listener\]**​ の右側にドラッグします。
2. プロパティウィンドウで、次の項目を設定します。
    
    <table class="tableblock frame-all grid-all fit-content spread">
<colgroup>
<col>
<col>
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">項目</th>
<th class="tableblock halign-left valign-top">値</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Display Name (表示名)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">ロガーの名前。</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Message (メッセージ)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>"New message arrived: "  payload  ", key:"  attributes.key  ", partition:"  attributes.partition  ", offset:" ++ attributes.offset</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock">Level (レベル)</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">INFO (Default) (INFO (デフォルト))</p></td>
</tr>
</tbody>
</table>
    
    次の図は、プロパティウィンドウでの ​**\[Logger\]**​ 設定の例を示しています。
    
    ![「Logger」 プロパティウィンドウ設定](/mulesoft/current/ja-jp/kafka-connector/4.6/_images/kafka-logger-config-consume.png)

### [](#この例の-xml)この例の XML

この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。

```xml
<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core"
xmlns:Kafka="http://www.mulesoft.org/schema/mule/Kafka"
	xmlns:http="http://www.mulesoft.org/schema/mule/http"
	xmlns="http://www.mulesoft.org/schema/mule/core"
	xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
	xmlns:spring="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/http
  http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
  http://www.springframework.org/schema/beans
	http://www.springframework.org/schema/beans/spring-beans-current.xsd
	http://www.mulesoft.org/schema/mule/core
	http://www.mulesoft.org/schema/mule/core/current/mule.xsd
	http://www.mulesoft.org/schema/mule/http
	http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
	http://www.mulesoft.org/schema/mule/ee/core
	http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd
	http://www.mulesoft.org/schema/mule/kafka
	http://www.mulesoft.org/schema/mule/kafka/current/mule-kafka.xsd">
  <configuration-properties file="mule-app.properties">
	</configuration-properties>
  <http:listener-config name="HTTP_Listener_config"
    doc:name="HTTP Listener config"  >
  <http:listener-connection host="0.0.0.0" port="8081" />
  </http:listener-config>

  <kafka:consumer-config name="Apache_Kafka_Consumer_configuration"
 	  doc:name="Apache Kafka Consumer configuration" >
	  <kafka:consumer-plaintext-connection groupId="${consumer.groupId}" >
	  <kafka:bootstrap-servers >
	    <kafka:bootstrap-server value="${config.basic.bootstrapServers}" />
	  </kafka:bootstrap-servers>
	  <kafka:topic-patterns >
	    <kafka:topic-pattern value='${topic1}
	    &lt;kafka:topic-pattern value="topic-1" /&gt;
	    &lt;kafka:topic-pattern value="topic-2" /&gt;
	    &lt;/kafka:topic-patterns&gt;]' />
	  </kafka:topic-patterns>
	  </kafka:consumer-plaintext-connection>
	</kafka:consumer-config>
	<kafka:producer-config name="Apache_Kafka_Producer_configuration"
	  doc:name="Apache Kafka Producer configuration" >
	  <kafka:producer-plaintext-connection >
	  <kafka:bootstrap-servers >
	    <kafka:bootstrap-server value="${config.basic.bootstrapServers}" />
	  </kafka:bootstrap-servers>
	  </kafka:producer-plaintext-connection>
	  </kafka:producer-config>
	<kafka:consumer-config name="Apache_Kafka_Consumer_configuration1" doc:name="Apache Kafka Consumer configuration" >
		<kafka:consumer-plaintext-connection groupId="${consumer.groupId}" >
			<kafka:bootstrap-servers >
				<kafka:bootstrap-server value="${config.basic.bootstrapServers}" />
			</kafka:bootstrap-servers>
			<kafka:topic-patterns >
				<kafka:topic-pattern value="${config.topics}" />
			</kafka:topic-patterns>
		</kafka:consumer-plaintext-connection>
	</kafka:consumer-config>
	<flow name="Producer-Flow" >
		<http:listener doc:name="Push message endpoint" config-ref="HTTP_Listener_config" path="/pushMessage" />
		<logger level="INFO" doc:name="Logger" message="&amp;quot;Message: '&amp;quot; ++ payload.message ++ &amp;quot;' is going to be published to topic: '&amp;quot; ++ payload.topic ++ &amp;quot;'.&amp;quot;" />
		<kafka:publish doc:name="Producer" topic="#[payload.topic]" key="#[now()]" config-ref="Apache_Kafka_Producer_configuration" />
		<set-payload value="Message successfully sent to Apache Kafka topic." doc:name="Push response builder" />
	</flow>
	<flow name="Consumer-Flow" >
		<kafka:message-listener doc:name="Consume message endpoint" config-ref="Apache_Kafka_Consumer_configuration"/>
		<logger level="INFO" doc:name="Logger" message="'New message arrived: ' ++ payload ++ &amp;quot;, key:&amp;quot; ++ attributes.key ++ &amp;quot;, partition:&amp;quot; ++ attributes.partition ++ &amp;quot;, offset:&amp;quot; ++ attributes.offset"/>
	</flow>
</mule>
```

## [](#関連情報)関連情報

- [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors)
- [MuleSoft ヘルプセンター](https://help.mulesoft.com)