進入正文，先用好 MSK 的部分。
雖然可以透過 aws cli 來部置 msk，但是如果情況允許，可以更直接的使用 Web 來進行快速部置。
1. 建立叢集，請選用自訂設定建立、打上叢集名稱、使用 MSK 預設組態。
2. 因為 MSK 必需要建立在 VPC 上的，所以若未有 VPC 請先按照上面的官方文件的第一、第二步先完成喔。
VPC 可以說是建立 MSK 最重要的一步，如果已經有會用到 Kafka 的 instance 存在，非常建議使用與該 intance 相同的 VPC，可以減少使用 VPC Peering Connection 的處理！
3. 代理程式可以先選最少資源使用量的，之後再增加代理程式的數目。
儲存體同理。
4. 加密可按需求選擇，但若沒有加密需求，建議只選統文字流量。下面的Spring Boot 例子會選用加密的方式開發，亦相對複雜一點。
5. 剩下的步驟可個人選填，然後點選建立叢集後就等吧，需要一點時間才能建好。若還沒建立 instance，現在就是個好時機先去把 instance 建好喔。
6. 當叢集建立完成後，我假設 instance 跟 msk 都在同一個 VPC 底下，這時我們可以看看官方文件的第五步，試著把 kafka Topic 建立起來。若然成功你就會看到Created topic XXXXXXXXX.。
當時我有遇到一個 Connection Time out 的問題，這問題出現的原因在於官方文件的第四步，設定 VPC 的安全群組上。我們需要把你會用來連上 MSK 的 instance 的安全群組加到 VPC 安全群組的傳入規則中，並設定為允許所有流量，這樣你的 client 才被允許連上 MSK 。
7. 當你確定建立 Topic 成功，恭喜你已經成功大半了！現在就來看 Spring Boot 部分的部置。
1.Spring Boot 先加上 Kafka 的 Dependency：
2. Kafka 的配置可以在 application.yml 中完成，為了簡化，我們直接用 yml 來完成所需要的 configuration。以下是我的範例結構：
有幾點要注意的：
1. security.protocol 若你叢集是選用加密，那要寫上 SSL，若選擇用純文字，那就要使用 PLAINTEXT。
2. ssl.trust-store-location 若是選用加密就請加上。這是是你的 truststor.jks 檔位置，我們可以透過找到自已使用的 JDK 資料夾位置，把 cacerts 這個檔案複制出來並重新命名為afka.client.truststore.jks 。
cp /usr/lib/jvm/<JDKFolder>/jre/lib/security/cacerts /tmp/kafka.client.truststore.jks如果今天是使用 Elastic Beanstalk 來進行部置，那我建議把 jks 檔放到你的 Resource 文件夾中，使用 classpath: 來讓 Kafka 讀取。這是跟 auto-scaling 有關，因為 Elastic Beanstalk 不會能你在 instance 中產生出來的文件一同複制到另一台 instance 中的。
如果是使用 EC2，那可以按照你有沒有AMI ／auto-scaling 等設定來決定你希望從哪邊獲得 jks 檔，但是你總要想辨法獲得 jks 檔。
3. bootstrap-servers 請放你的 MSK BrokerString。這可以在你的 MSK 叢集管理頁，檢視用戶端資訊中獲的。
你會拿到一大串字串，請直接複制貼上到 yml 的producer 與 consumer 中的 bootstrap-servers 後即可。
4. group-id 可以打你任意需要的字串。
5. 在 Controller 與 Service 中分別寫上 Kafka 的生產與消費邏輯。
要接收 Kafka 的消息，我們可以使用 @KafkaListener，後面的 topics 就是當時你創立的 topics 名稱囉。
6. 試著上傳到 instance 然後跑起來吧！
如果你覺得我的文章幫助到你，希望你也可以為文章拍手，分別 Follow 我的個人頁與程式愛好者出版，按讚我們的粉絲頁喔，支持我們推出更多更好的內容創作！
