尽管微服务架构可能并不是所有系统的灵丹妙药，但它无疑具有其优势，尤其是在构建具有许多不同组件的复杂系统时。如果您正在考虑微服务，则考虑不同服务的通信方式。在本文中，我们将研究如何设置Apache Kafka实例，创建用户服务以将数据发布到主题，以及构建通知服务以使用这些主题中的数据。具体来说，我们将构建一个两步验证应用程序，用户可以在该程序中进行注册，接收带有验证码的邮件，并使用该代码完成注册。源代码可以在这里找到。
为什么选择Apache Kafka？
Kafka是LinkedIn于2011年创建的分布式流媒体平台，用于处理高吞吐量，低延迟传输以及实时处理记录流。它的三大功能使其非常适合此用例：
- 发布和订阅记录流。在这方面，它类似于消息队列或企业消息传递系统。
- 以容错方式存储记录流。
- 处理发生的记录流。
设置Apache Kafka
开始本教程之前，需要满足以下条件：
- Mac版Docker或Windows版Docker
- Docker Compose的知识
- Node.js的知识
我们将使用Wurstmeister Kafka Docker映像。请注意，Kafka使用Zookeeper在不同的Kafka节点之间进行协调。
docker-compose.yml类似docker-compose.yml用于为Kafka和Zookeeper提取图像。Kafka服务所需的配置选项之一是KAFKA_ZOOKEEPER_CONNECT ，它告诉Kafka在哪里可以找到Zookeeper实例。
1. version: '2.1'
2. services:
3. zookeeper:
4. container_name: zookeeper
5. image: wurstmeister/zookeeper
6. ports:
7. - "2181:2181"
8. kafka:
9. container_name: kafka
10. image: wurstmeister/kafka
11. ports:
12. - "9092"
13. depends_on:
14. - "zookeeper"
15. environment:
16. KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
将数据发布到Kafka主题
要将数据发布到Kafka主题，我们将创建一个提供两个端点的用户服务：
/api/register –将用户详细信息存储在内存中存储节点缓存中，并将用户数据发布到Kafka主题user_account_created 。
/api/verify –验证提供的代码正确，并将用户数据发布到Kafka主题user_account_verified 。
我们使用node-rdkafka NPM软件包创建一个生产者，该生产者从我们的节点应用程序连接到Kafka：
1. let producerReady;
2. producer = new kafka.Producer({
3. debug: 'all',
4. 'client.id': 'user-api',
5. 'metadata.broker.list': KAFKA_BROKER_LIST,
6. 'compression.codec': 'gzip',
7. 'retry.backoff.ms': 200,
8. 'message.send.max.retries': 10,
9. 'socket.keepalive.enable': true,
10. 'queue.buffering.max.messages': 100000,
11. 'queue.buffering.max.ms': 1000,
12. 'batch.num.messages': 1000000,
13. dr_cb: true
14. });
15. producer.connect({}, err => {
16. if (err) {
17. logger.error('connect', err);
18. }
19. });
20. producerReady = new Promise((resolve, reject) => {
21. producer.on('ready', () => {
22. logger.info('producer ready');
23. resolve(producer);
24. });
25. });我们创建一个新的Promise对象，该对象解析为准备开始发布数据的生产者。这在我们的sendMessage函数中使用，该函数将数据发布到Kafka主题分区：
1. KafkaService.prototype.sendMessage = function sendMessage(
2. topic,
3. payload,
4. partition = 0
5. ) {
6. return producerReady
7. .then(producer => {
8. const message = Buffer.from(JSON.stringify(payload));
9. producer.produce(topic, partition, message);
10. })
11. .catch(error => logger.error('unable to send message', error));
12. };使用Kafka主题中的数据
为了使用Kafka主题中的数据，我们将创建一个通知服务，以监听来自我们主题的数据，并根据从中获取数据的主题发送带有验证码或成功消息的电子邮件。
我们创建一个连接到Kafka的使用者，其中KAFKA_BROKER_LIST是所有Kafka实例的逗号分隔列表。
1. process.stdin.resume(); // keep process alive
2.
3. require('dotenv').config();
4.
5. const Kafka = require('node-rdkafka');
6.
7. const logger = require('./logger');
8.
9. const sendMail = require('./email');
10.
11. const KAFKA_BROKER_LIST = process.env.KAFKA_BROKER_LIST;
12.
13. const consumer = new Kafka.KafkaConsumer({
14. //'debug': 'all',
15. 'metadata.broker.list': KAFKA_BROKER_LIST,
16. 'group.id': 'notification-service',
17. 'enable.auto.commit': false
18. });node-rdkafka返回的使用者对象是可读流的实例。我们等待ready事件订阅我们的主题user_account_created和user_account_verified ，并侦听这些主题中的数据：
1. const topics = [
2. 'user_account_created',
3. 'user_account_verified'
4. ];
5.
6. //counter to commit offsets every numMessages are received
7. let counter = 0;
8. let numMessages = 5;
9.
10. consumer.on('ready', function(arg) {
11. logger.info('consumer ready.' + JSON.stringify(arg));
12.
13. consumer.subscribe(topics);
14. //start consuming messages
15. consumer.consume();
16. });
17.
18. consumer.on('data', function(metadata) {
19. counter++;
20.
21. //committing offsets every numMessages
22. if (counter % numMessages === 0) {
23. logger.info('calling commit');
24. consumer.commit(metadata);
25. }
26.
27. // Output the actual message contents
28. const data = JSON.parse(metadata.value.toString());
29. logger.info('data value', data);
30.
31. if(metadata.topic === 'user_account_created'){
32. const to = data.email;
33. const subject = 'Verify Account';
34. const content = `Hello ${data.first_name},
35. Please use this code ${data.code} to complete your verification`;
36. sendMail(subject, content,to);
37. }else if(metadata.topic === 'user_account_verified') {
38. const to = data.email;
39. const subject = 'Account Verified';
40. const content = `Hello ${data.first_name},
41. You have successfully been verified`;
42. sendMail(subject, content,to);
43. }
44.
45. });
46.
47. consumer.on('disconnected', function(arg) {
48. logger.info('consumer disconnected. ' + JSON.stringify(arg));
49. });
50.
51. //logging all errors
52. consumer.on('event.error', function(err) {
53. logger.error('Error from consumer', err, 'code: ', err.code);
54. });
55.
56. //starting the consumer
57. consumer.connect();当消息发布到我们正在侦听的任何主题时，将调用data事件处理程序。在这里，我们解析传入的消息并检查元数据对象，以了解接收到的数据所针对的主题，因此我们可以执行适当的操作。
结论
我们的两因素身份验证应用程序演示了使用Apache Kafka（还有其他系统，例如RabbitMQ ， ZeroMQ ）的两个微服务之间的通信模式，相对于Feign增加了未来的灵活性。
例如，假设我们将来会添加一个推荐服务，当新用户登录时，该服务需要发送推荐；它仅订阅user_account_verified主题，因此无需更改用户服务。
系统架构结论（1）数据管道场景，MQ比cache更合适；（2）服务化架构，不应该绕过service读取其后端的cache/db，而应该通过RPC接口访问；（3）MQ通信可扩展
