事件消费者
调整 service/domain/model/trend.py:
@@ -9,4 +9,3 @@ from .book import Book
class Trend:
query: str
books: List[Book]
- created_at: datetime | None
可以使用 kafka 事件的时间戳来进行时间追踪，此处可以移去时间字段。
调整 service/trend/domain/gateway/trend_manager.py:
@@ -1,8 +1,10 @@
from abc import ABC, abstractmethod
-from typing import List
+from typing import Callable, List
from ....domain.model import Trend
+ConsumeCallback = Callable[[bytes, bytes], None]
+
class TrendManager(ABC):
@abstractmethod
@@ -10,5 +12,15 @@ class TrendManager(ABC):
pass
@abstractmethod
- def top_trends(self, offset: int) -> List[Trend]:
+ def top_trends(self, page_size: int) -> List[Trend]:
+ pass
+
+
+class TrendEventConsumer(ABC):
+ @abstractmethod
+ def consume_events(self, callback: ConsumeCallback):
+ pass
+
+ @abstractmethod
+ def stop(self):
pass
实现 TrendEventConsumer，service/trend/infrastructure/mq/kafka.py:
import sys
from typing import List
from confluent_kafka import Consumer, KafkaException
from ...domain.gateway import TrendEventConsumer, ConsumeCallback
GROUP_ID = "trend-svr"
class KafkaConsumer(TrendEventConsumer):
def __init__(self, brokers: List[str], topic: str):
self.consumer = Consumer(
{'bootstrap.servers': ','.join(brokers),
'group.id': GROUP_ID,
'auto.offset.reset': 'smallest'})
self.topic = topic
self.running = False
def consume_events(self, callback: ConsumeCallback):
try:
self.consumer.subscribe([self.topic])
self.running = True
while self.running:
msg = self.consumer.poll(timeout=1.0) # Poll for messages
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaException._PARTITION_EOF:
# End of partition
sys.stderr.write('%% {} [{}] reached end at offset {} - {}\n'.format(
msg.topic(), msg.partition(), msg.offset()))
elif msg.error():
raise KafkaException(msg.error())
else:
# Process message
callback(msg.key(), msg.value())
finally:
self.consumer.close()
def stop(self):
self.running = False
调整 TrendManager 实现，service/trend/infrastructure/cache/redis.py:
@@ -29,13 +29,13 @@ class RedisCache(TrendManager):
self.c.set(k, results)
return score
- def top_trends(self, offset: int) -> List[Trend]:
+ def top_trends(self, page_size: int) -> List[Trend]:
top_items: Any = self.c.zrevrange(
- trends_key, 0, offset, withscores=True)
+ trends_key, 0, page_size - 1, withscores=True)
trends = []
for item in top_items:
query = item[0]
- t = Trend(query=query, books=[], created_at=None)
+ t = Trend(query=query, books=[])
k = query_key_prefix + query
value: Any = self.c.get(k)
if value is not None:
调整 service/trend/application/wire_helper.py:
@@ -1,17 +1,23 @@
-from ..domain.gateway import TrendManager
+from ..domain.gateway import TrendManager, TrendEventConsumer
from ..infrastructure.config import Config
from ..infrastructure.cache import RedisCache
+from ..infrastructure.mq import KafkaConsumer
class WireHelper:
- def __init__(self, kvStore: RedisCache):
+ def __init__(self, kvStore: RedisCache, consumer: KafkaConsumer):
self.kvStore = kvStore
+ self.consumer = consumer
@classmethod
def new(cls, c: Config):
kv = RedisCache(c.cache.host, c.cache.port,
c.cache.password, c.cache.db)
- return cls(kv)
+ consumer = KafkaConsumer(c.mq.brokers, c.mq.topic)
+ return cls(kv, consumer)
def trend_manager(self) -> TrendManager:
return self.kvStore
+
+ def trend_event_consumer(self) -> TrendEventConsumer:
+ return self.consumer
添加配置项，service/trend/infrastructure/config/config.py:
@@ -1,8 +1,15 @@
from dataclasses import dataclass
+from typing import List
import yaml
+@dataclass
+class MQConfig:
+ brokers: List[str]
+ topic: str
+
+
@dataclass
class CacheConfig:
host: str
@@ -14,11 +21,14 @@ class CacheConfig:
@dataclass
class Config:
cache: CacheConfig
+ mq: MQConfig
def parseConfig(filename: str) -> Config:
with open(filename, 'r') as f:
data = yaml.safe_load(f)
return Config(
- CacheConfig(**data['cache'])
+ CacheConfig(**data['cache']),
+ MQConfig(**data['mq'])
+
)
填入配置值，service/web/config.yml:
@@ -3,3 +3,7 @@ cache:
port: 6379
password: test_pass
db: 0
+mq:
+ brokers:
+ - localhost:9094
+ topic: lr-book-searches
此处使用
9094端口是为了从外部访问跑在 docker 容器内的 kafka。
详情阅读 “Accessing Apache Kafka with internal and external clients“配置一节，bitnami/kafka镜像。
添加 service/trend/application/consumer/trend.py:
import json
from ....domain.model import Trend
from ...domain.gateway import TrendManager, TrendEventConsumer
class TrendConsumer():
def __init__(self, trend_manager: TrendManager, event_consumer: TrendEventConsumer):
self.trend_manager = trend_manager
self.event_consumer = event_consumer
def start(self):
def process_event(key: bytes, data: bytes):
t = Trend(query=key.decode('utf-8'), books=json.loads(data))
self.trend_manager.create_trend(t)
self.event_consumer.consume_events(process_event)
def get_event_consumer(self) -> TrendEventConsumer:
return self.event_consumer
调整 service/trend/application/executor/trend_operator.py:
@@ -9,8 +9,5 @@ class TrendOperator():
def __init__(self, trend_manager: TrendManager):
self.trend_manager = trend_manager
- def create_trend(self, t: Trend) -> int:
- return self.trend_manager.create_trend(t)
-
- def top_trends(self, offset: int) -> List[Trend]:
- return self.trend_manager.top_trends(offset)
+ def top_trends(self, page_size: int) -> List[Trend]:
+ return self.trend_manager.top_trends(page_size)
调整 service/trend/adapter/router.py:
@@ -11,9 +11,9 @@ class RestHandler:
self._logger = logger
self.trend_operator = trend_operator
- def get_trends(self, offset: int):
+ def get_trends(self, page_size: int):
try:
- return self.trend_operator.top_trends(offset)
+ return self.trend_operator.top_trends(page_size)
except Exception as e:
self._logger.error(f"Failed to get trends: {e}")
raise HTTPException(status_code=404, detail="Failed to get trends")
@@ -26,5 +26,5 @@ def make_router(app: FastAPI, wire_helper: WireHelper):
)
@app.get("/trends")
- async def get_trends(o: int = 0):
- return rest_handler.get_trends(o)
+ async def get_trends(ps: int = 10):
+ return rest_handler.get_trends(ps)
修改 main.py，使得 consumer 和 router 都能优雅关闭：
@@ -1,12 +1,35 @@
+from contextlib import asynccontextmanager
+import threading
+
from fastapi import FastAPI
from .adapter.router import make_router
from .application import WireHelper
+from .application.consumer import TrendConsumer
from .infrastructure.config import parseConfig
CONFIG_FILENAME = "service/trend/config.yml"
c = parseConfig(CONFIG_FILENAME)
wire_helper = WireHelper.new(c)
-app = FastAPI()
+
+# Run the consumer
+tc = TrendConsumer(wire_helper.trend_manager(),
+ wire_helper.trend_event_consumer())
+event_consumer = tc.get_event_consumer()
+consumer_thread = threading.Thread(target=tc.start)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+ # Run at startup
+ consumer_thread.start()
+ yield
+ # Run on shutdown
+ event_consumer.stop()
+ consumer_thread.join()
+ print("Event consumer is closed! Bye!")
+
+# Run the FastAPI server
+app = FastAPI(lifespan=lifespan)
make_router(app, wire_helper)
Kafka 消费者和 FastAPI 服务器是运行在同一个进程中的 2 个线程。这里使用 FastAPI 的 lifespan 函数来优雅地关闭二者。
提示: 你可以尝试使用
lifespan函数来解决 web 服务关闭时候容易出现的如下问题：
“Producer terminating with 1 message (248 bytes) still in queue or transit: use flush() to wait for outstanding message delivery“
记得创建或修改各种 __.init__.py 文件来优化符号导出。
- service/trend/domain/gateway/__init__.py
- service/trend/application/consumer/__init__.py
- service/trend/infrastructure/mq/__init__.py
启动 Web 服务和热搜服务。尝试在 Web 服务的首页 http://localhost:8000/ 中搜索关键词，如“love”、“war” 和 “wealth”。
然后你应该能够从热搜服务的网址 (http://localhost:8001/trends) 中看到类似以下的一些结果：
[
{
"query": "love",
"books": [
{
"id": 4,
"title": "Pride and Prejudice",
"author": "Jane Austen",
"published_at": "1813-01-28",
"description": "A classic novel exploring the themes of love, reputation, and social class in Georgian England.",
"created_at": "2024-04-02T21:02:59.314+08:00"
},
{
"id": 10,
"title": "War and Peace",
"author": "Leo Tolstoy",
"published_at": "1869-01-01",
"description": "A novel depicting the Napoleonic era in Russia, exploring themes of love, war, and historical determinism.",
"created_at": "2024-04-02T21:02:59.42+08:00"
}
]
},
{
"query": "war",
"books": [
{
"id": 10,
"title": "War and Peace",
"author": "Leo Tolstoy",
"published_at": "1869-01-01",
"description": "A novel depicting the Napoleonic era in Russia, exploring themes of love, war, and historical determinism.",
"created_at": "2024-04-02T21:02:59.42+08:00"
},
{
"id": 12,
"title": "The Odyssey",
"author": "Homer",
"published_at": "8th Century BC",
"description": "An ancient Greek epic poem attributed to Homer, detailing the journey of Odysseus after the Trojan War.",
"created_at": "2024-04-02T21:02:59.455+08:00"
}
]
},
{
"query": "wealth",
"books": [
{
"id": 1,
"title": "The Great Gatsby",
"author": "F. Scott Fitzgerald",
"published_at": "1925-04-10",
"description": "A novel depicting the opulent lives of wealthy Long Island residents during the Jazz Age.",
"created_at": "2024-04-02T20:36:01.015+08:00"
}
]
}
]
Loading...
> 此处输出代码运行结果