Microservices Communication Patterns: A Comprehensive Guide#

1
sequenceDiagram
2
    participant Client
3
    participant API Gateway
4
    participant Order Service
5
    participant Inventory Service
6
    participant Payment Service
7

8
    Client->>API Gateway: POST /orders
9
    API Gateway->>Order Service: Create Order
10

11
    Order Service->>Inventory Service: Check Stock
12
    Inventory Service-->>Order Service: Stock Available
13

14
    Order Service->>Payment Service: Process Payment
15
    Payment Service-->>Order Service: Payment Confirmed
16

17
    Order Service-->>API Gateway: Order Created
18
    API Gateway-->>Client: 201 Created

1
// Order Service
2
app.post("/orders", async (req, res) => {
3
  try {
4
    // Check inventory
5
    const stockResponse = await fetch("http://inventory-service/check", {
6
      method: "POST",
7
      body: JSON.stringify({ items: req.body.items }),
8
      headers: { "Content-Type": "application/json" },
9
    });
10

11
    if (!stockResponse.ok) {
12
      return res.status(400).json({ error: "Insufficient stock" });
13
    }
14

15
    // Process payment
16
    const paymentResponse = await fetch("http://payment-service/charge", {
17
      method: "POST",
18
      body: JSON.stringify({
19
        amount: req.body.total,
20
        customerId: req.body.customerId,
21
      }),
22
      headers: { "Content-Type": "application/json" },
23
    });
24

25
    if (!paymentResponse.ok) {
26
      return res.status(400).json({ error: "Payment failed" });
27
    }
28

29
    // Create order
30
    const order = await createOrder(req.body);
31
    res.status(201).json(order);
32
  } catch (error) {
33
    res.status(500).json({ error: "Internal server error" });
34
  }
35
});

1
graph LR
2
    subgraph "GraphQL Gateway"
3
        Schema[GraphQL Schema]
4
        Resolver[Resolvers]
5
    end
6

7
    Client[Client App] -->|Query| Schema
8
    Schema --> Resolver
9

10
    Resolver --> UserService[User Service]
11
    Resolver --> OrderService[Order Service]
12
    Resolver --> ProductService[Product Service]
13

14
    UserService -->|User Data| Resolver
15
    OrderService -->|Order Data| Resolver
16
    ProductService -->|Product Data| Resolver
17

18
    Resolver -->|Combined Response| Client

1
sequenceDiagram
2
    participant Client
3
    participant Server
4

5
    Note over Client,Server: Bidirectional Streaming
6

7
    Client->>Server: Stream Request 1
8
    Client->>Server: Stream Request 2
9
    Server->>Client: Stream Response 1
10
    Client->>Server: Stream Request 3
11
    Server->>Client: Stream Response 2
12
    Server->>Client: Stream Response 3
13

14
    Note over Client,Server: Connection remains open

1
syntax = "proto3";
2

3
service OrderService {
4
    // Unary RPC
5
    rpc CreateOrder(OrderRequest) returns (OrderResponse);
6

7
    // Server streaming RPC
8
    rpc ListOrders(ListOrdersRequest) returns (stream Order);
9

10
    // Client streaming RPC
11
    rpc UploadOrders(stream Order) returns (UploadSummary);
12

13
    // Bidirectional streaming RPC
14
    rpc ProcessOrders(stream OrderRequest) returns (stream OrderStatus);
15
}
16

17
message OrderRequest {
18
    string customer_id = 1;
19
    repeated OrderItem items = 2;
20
    double total_amount = 3;
21
}
22

23
message OrderResponse {
24
    string order_id = 1;
25
    string status = 2;
26
    int64 created_at = 3;
27
}

1
graph LR
2
    subgraph "Message Queue System"
3
        Queue1[Order Queue]
4
        Queue2[Email Queue]
5
        Queue3[Analytics Queue]
6
        DLQ[Dead Letter Queue]
7
    end
8

9
    OrderService[Order Service] -->|Publish| Queue1
10
    Queue1 -->|Consume| PaymentService[Payment Service]
11
    Queue1 -->|Failed Messages| DLQ
12

13
    PaymentService -->|Publish| Queue2
14
    Queue2 -->|Consume| EmailService[Email Service]
15

16
    OrderService -->|Publish| Queue3
17
    Queue3 -->|Consume| AnalyticsService[Analytics Service]
18

19
    DLQ -->|Retry/Alert| Monitor[Monitoring System]

1
// Publisher
2
const amqp = require("amqplib");
3

4
async function publishOrder(order) {
5
  const connection = await amqp.connect("amqp://localhost");
6
  const channel = await connection.createChannel();
7

8
  await channel.assertQueue("order_queue", { durable: true });
9

10
  const message = Buffer.from(JSON.stringify(order));
11
  channel.sendToQueue("order_queue", message, { persistent: true });
12

13
  console.log(" [x] Sent order:", order.id);
14
  await channel.close();
15
  await connection.close();
16
}
17

18
// Consumer
19
async function consumeOrders() {
20
  const connection = await amqp.connect("amqp://localhost");
21
  const channel = await connection.createChannel();
22

23
  await channel.assertQueue("order_queue", { durable: true });
24
  channel.prefetch(1); // Process one message at a time
25

26
  console.log(" [*] Waiting for orders...");
27

28
  channel.consume("order_queue", async msg => {
29
    const order = JSON.parse(msg.content.toString());
30

31
    try {
32
      await processOrder(order);
33
      channel.ack(msg); // Acknowledge successful processing
34
    } catch (error) {
35
      console.error("Processing failed:", error);
36
      channel.nack(msg, false, false); // Send to DLQ
37
    }
38
  });
39
}

1
graph TB
2
    subgraph "Event Bus"
3
        Topic1[Order Events]
4
        Topic2[User Events]
5
        Topic3[Product Events]
6
    end
7

8
    OrderService[Order Service] -->|Publish OrderCreated| Topic1
9
    UserService[User Service] -->|Publish UserRegistered| Topic2
10
    ProductService[Product Service] -->|Publish ProductUpdated| Topic3
11

12
    Topic1 -->|Subscribe| EmailService[Email Service]
13
    Topic1 -->|Subscribe| InventoryService[Inventory Service]
14
    Topic1 -->|Subscribe| AnalyticsService[Analytics Service]
15

16
    Topic2 -->|Subscribe| EmailService
17
    Topic2 -->|Subscribe| RecommendationService[Recommendation Service]
18

19
    Topic3 -->|Subscribe| CacheService[Cache Service]
20
    Topic3 -->|Subscribe| SearchService[Search Service]

1
graph LR
2
    subgraph "Kafka Cluster"
3
        subgraph "Topics"
4
            T1[orders]
5
            T2[payments]
6
            T3[inventory]
7
        end
8

9
        subgraph "Partitions"
10
            P1[Partition 0]
11
            P2[Partition 1]
12
            P3[Partition 2]
13
        end
14

15
        T1 --> P1
16
        T1 --> P2
17
        T1 --> P3
18
    end
19

20
    Producer1[Order Service] -->|Produce| T1
21
    Producer2[Payment Service] -->|Produce| T2
22

23
    subgraph "Consumer Group A"
24
        C1[Consumer 1] -->|Read| P1
25
        C2[Consumer 2] -->|Read| P2
26
        C3[Consumer 3] -->|Read| P3
27
    end
28

29
    subgraph "Consumer Group B"
30
        C4[Analytics Consumer] -->|Read All| T1
31
    end
32

33
    T1 -->|Stream| StreamProcessor[Stream Processor]
34
    StreamProcessor -->|Enriched Events| T3

1
// Producer
2
const { Kafka } = require("kafkajs");
3

4
const kafka = new Kafka({
5
  clientId: "order-service",
6
  brokers: ["localhost:9092"],
7
});
8

9
const producer = kafka.producer();
10

11
async function publishOrderEvent(order) {
12
  await producer.connect();
13

14
  await producer.send({
15
    topic: "orders",
16
    messages: [
17
      {
18
        key: order.customerId,
19
        value: JSON.stringify({
20
          eventType: "OrderCreated",
21
          orderId: order.id,
22
          customerId: order.customerId,
23
          amount: order.amount,
24
          timestamp: Date.now(),
25
        }),
26
        headers: {
27
          "correlation-id": order.correlationId,
28
        },
29
      },
30
    ],
31
  });
32

33
  await producer.disconnect();
34
}
35

36
// Consumer
37
const consumer = kafka.consumer({ groupId: "payment-service" });
38

39
async function consumeOrderEvents() {
40
  await consumer.connect();
41
  await consumer.subscribe({ topic: "orders", fromBeginning: false });
42

43
  await consumer.run({
44
    eachMessage: async ({ topic, partition, message }) => {
45
      const event = JSON.parse(message.value.toString());
46

47
      console.log({
48
        topic,
49
        partition,
50
        offset: message.offset,
51
        event,
52
      });
53

54
      if (event.eventType === "OrderCreated") {
55
        await processPayment(event);
56
      }
57
    },
58
  });
59
}

1
GET /api/v1/users
2
GET /api/v2/users

1
GET /api/users
2
Accept-Version: v1

1
GET /api/users
2
Accept: application/vnd.myapp.v1+json

1
GET /api/users?version=1

1
graph TD
2
    Start[Communication Need] --> Immediate{Need Immediate Response?}
3

4
    Immediate -->|Yes| Sync[Synchronous]
5
    Immediate -->|No| Async[Asynchronous]
6

7
    Sync --> DataReq{Complex Data Requirements?}
8
    DataReq -->|Yes| GraphQL[Use GraphQL]
9
    DataReq -->|No| Performance{High Performance Critical?}
10

11
    Performance -->|Yes| gRPC[Use gRPC]
12
    Performance -->|No| REST[Use REST]
13

14
    Async --> Volume{High Volume?}
15
    Volume -->|Yes| Streaming{Need Replay?}
16
    Volume -->|No| Queue[Use Message Queue]
17

18
    Streaming -->|Yes| Kafka[Use Kafka]
19
    Streaming -->|No| PubSub[Use Pub/Sub]

1
class CircuitBreaker {
2
  constructor(threshold = 5, timeout = 60000) {
3
    this.threshold = threshold;
4
    this.timeout = timeout;
5
    this.failures = 0;
6
    this.state = "CLOSED";
7
    this.nextAttempt = Date.now();
8
  }
9

10
  async call(fn) {
11
    if (this.state === "OPEN") {
12
      if (Date.now() < this.nextAttempt) {
13
        throw new Error("Circuit breaker is OPEN");
14
      }
15
      this.state = "HALF_OPEN";
16
    }
17

18
    try {
19
      const result = await fn();
20
      this.onSuccess();
21
      return result;
22
    } catch (error) {
23
      this.onFailure();
24
      throw error;
25
    }
26
  }
27

28
  onSuccess() {
29
    this.failures = 0;
30
    this.state = "CLOSED";
31
  }
32

33
  onFailure() {
34
    this.failures++;
35
    if (this.failures >= this.threshold) {
36
      this.state = "OPEN";
37
      this.nextAttempt = Date.now() + this.timeout;
38
    }
39
  }
40
}

1
async function retryWithExponentialBackoff(fn, maxRetries = 3) {
2
  let lastError;
3

4
  for (let i = 0; i < maxRetries; i++) {
5
    try {
6
      return await fn();
7
    } catch (error) {
8
      lastError = error;
9
      const delay = Math.min(1000 * Math.pow(2, i), 10000);
10
      await new Promise(resolve => setTimeout(resolve, delay));
11
    }
12
  }
13

14
  throw lastError;
15
}

1
async function callWithTimeout(fn, timeout = 5000) {
2
  const timeoutPromise = new Promise((_, reject) => {
3
    setTimeout(() => reject(new Error("Timeout")), timeout);
4
  });
5

6
  return Promise.race([fn(), timeoutPromise]);
7
}