chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,507 @@
|
||||
# NATS EventBus Integration Guide
|
||||
|
||||
## Overview
|
||||
|
||||
This document provides a comprehensive guide for integrating NATS as an EventBus in Coze Studio, including architecture design, implementation details, configuration instructions, and usage guidelines.
|
||||
|
||||
## Integration Background
|
||||
|
||||
### Why Choose NATS?
|
||||
|
||||
In Coze Studio's architecture, EventBus plays a critical role in asynchronous message delivery, including workflow execution, Agent communication, data processing pipelines, and other core functions. NATS, as a lightweight and high-performance messaging system, brings the following core advantages to Coze Studio:
|
||||
|
||||
1. **Lightweight**: NATS has minimal resource footprint and simple deployment architecture, perfect for cloud-native environments
|
||||
2. **High Performance**: Provides low-latency, high-throughput messaging that can support Coze Studio's large-scale concurrent Agent execution
|
||||
3. **Simplicity**: Clean and intuitive API that reduces development and maintenance costs
|
||||
4. **JetStream Support**: Provides message persistence, replay, and stream processing capabilities through JetStream
|
||||
5. **Cloud Native**: Native support for Kubernetes, easy to deploy and manage in containerized environments
|
||||
6. **Security**: Built-in authentication and authorization mechanisms with TLS encryption support
|
||||
|
||||
### Comparison with Other MQ Systems
|
||||
|
||||
| Feature | NATS | NSQ | Kafka | RocketMQ | Pulsar |
|
||||
| ---------------------- | -------------- | -------------- | -------------- | -------------- | -------------- |
|
||||
| **Deployment Complexity** | Very Low | Low | Medium | Medium | Medium |
|
||||
| **Performance** | Very High | Medium | High | High | High |
|
||||
| **Resource Usage** | Very Low | Low | Medium | Medium | Medium |
|
||||
| **Message Persistence** | JetStream | Limited | Strong | Strong | Strong |
|
||||
| **Message Ordering** | Supported | Weak | Strong | Strong | Strong |
|
||||
| **Horizontal Scaling** | Good | Medium | Good | Good | Excellent |
|
||||
| **Operational Complexity** | Very Low | Low | High | Medium | Medium |
|
||||
| **Cloud Native Support** | Excellent | Medium | Medium | Medium | Good |
|
||||
|
||||
#### NATS Core Advantages
|
||||
|
||||
**Lightweight and High Performance**:
|
||||
- **Memory Usage**: NATS server typically requires only tens of MB to handle millions of messages
|
||||
- **Startup Speed**: Second-level startup, perfect for microservices and containerized deployments
|
||||
- **Latency**: Sub-millisecond message latency, suitable for real-time scenarios
|
||||
- **Throughput**: Single node can handle millions of messages per second
|
||||
|
||||
**Simplicity**:
|
||||
- **Simple Configuration**: Minimal configuration required to run, no complex cluster setup needed
|
||||
- **Clean API**: Publish/subscribe pattern is simple and intuitive with low learning curve
|
||||
- **Operations Friendly**: Rich monitoring and debugging tools, easy troubleshooting
|
||||
|
||||
**Cloud Native Features**:
|
||||
- **Kubernetes Integration**: Official Helm Charts and Operators available
|
||||
- **Service Discovery**: Built-in service discovery mechanism, no external dependencies
|
||||
- **Elastic Scaling**: Supports dynamic cluster membership changes
|
||||
|
||||
## Architecture Design
|
||||
|
||||
### Overall Architecture
|
||||
|
||||
```
|
||||
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
|
||||
│ Coze Studio │ │ NATS Server │ │ JetStream │
|
||||
│ Application │ │ │ │ Storage │
|
||||
├─────────────────┤ ├─────────────────┤ ├─────────────────┤
|
||||
│ Producer │───▶│ Core NATS │ │ Streams │
|
||||
│ Consumer │◀───│ JetStream │◀───│ Consumers │
|
||||
│ EventBus │ │ Clustering │ │ Key-Value │
|
||||
└─────────────────┘ └─────────────────┘ └─────────────────┘
|
||||
```
|
||||
|
||||
### Message Flow Patterns
|
||||
|
||||
NATS supports two messaging modes in Coze Studio:
|
||||
|
||||
1. **Core NATS**: For real-time, lightweight message delivery
|
||||
- Publish/Subscribe pattern
|
||||
- Request/Response pattern
|
||||
- Queue Group pattern
|
||||
|
||||
2. **JetStream**: For messages requiring persistence and high reliability
|
||||
- Stream storage
|
||||
- Message replay
|
||||
- Consumer acknowledgment mechanism
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### Producer Implementation
|
||||
|
||||
The Producer is responsible for sending messages to NATS, supporting the following features:
|
||||
|
||||
```go
|
||||
type Producer struct {
|
||||
nc nats.Conn
|
||||
js nats.JetStreamContext
|
||||
closed bool
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
func (p *Producer) SendMessage(ctx context.Context, topic string, message []byte) error {
|
||||
// Supports both Core NATS and JetStream modes
|
||||
if p.js != nil {
|
||||
// JetStream mode: supports message persistence
|
||||
_, err := p.js.Publish(topic, message)
|
||||
return err
|
||||
} else {
|
||||
// Core NATS mode: lightweight publishing
|
||||
return p.nc.Publish(topic, message)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Consumer Implementation
|
||||
|
||||
The Consumer is responsible for receiving and processing messages from NATS:
|
||||
|
||||
```go
|
||||
func (c *Consumer) RegisterConsumer(serverURL, topic, group string, handler ConsumerHandler) error {
|
||||
// Choose JetStream or Core NATS based on configuration
|
||||
if c.useJetStream {
|
||||
return c.startJetStreamConsumer(ctx, topic, group, handler)
|
||||
} else {
|
||||
return c.startCoreConsumer(ctx, topic, group, handler)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### JetStream Consumer Features
|
||||
|
||||
- **Message Acknowledgment**: Supports manual acknowledgment mechanism to ensure successful message processing
|
||||
- **Retry Mechanism**: Automatic retry for failed messages with exponential backoff support
|
||||
- **Sequential Processing**: Single message processing to avoid complexity from batch processing
|
||||
- **Flow Control**: Precise message flow control to prevent consumer overload
|
||||
|
||||
#### Core NATS Consumer Features
|
||||
|
||||
- **Queue Groups**: Supports load-balanced message distribution
|
||||
- **Lightweight**: No persistence overhead, suitable for real-time message processing
|
||||
- **High Performance**: Extremely low message processing latency
|
||||
|
||||
## Configuration Guide
|
||||
|
||||
### Environment Variables
|
||||
|
||||
Add the following NATS-related configurations in `docker/.env.example`:
|
||||
|
||||
```bash
|
||||
# Backend Event Bus
|
||||
export COZE_MQ_TYPE="nats" # Set message queue type to NATS
|
||||
export MQ_NAME_SERVER="nats:4222" # NATS server address
|
||||
|
||||
# NATS specific configuration
|
||||
# NATS_SERVER_URL: NATS server connection URL, supports nats:// and tls:// protocols
|
||||
# For cluster setup, use comma-separated URLs: "nats://nats1:4222,nats://nats2:4222"
|
||||
# For TLS connection: "tls://nats:4222"
|
||||
export NATS_SERVER_URL="nats://nats:4222"
|
||||
|
||||
# NATS_JWT_TOKEN: JWT token for NATS authentication (leave empty for no auth)
|
||||
export NATS_JWT_TOKEN=""
|
||||
|
||||
# NATS_NKEY_SEED: Path to NATS seed file for NKey authentication (optional)
|
||||
export NATS_NKEY_SEED=""
|
||||
|
||||
# NATS_USERNAME: Username for NATS authentication (optional)
|
||||
export NATS_USERNAME=""
|
||||
|
||||
# NATS_PASSWORD: Password for NATS authentication (optional)
|
||||
export NATS_PASSWORD=""
|
||||
|
||||
# NATS_TOKEN: Token for NATS authentication (optional)
|
||||
export NATS_TOKEN=""
|
||||
|
||||
# NATS_STREAM_REPLICAS: Number of replicas for JetStream streams (default: 1)
|
||||
export NATS_STREAM_REPLICAS="1"
|
||||
|
||||
# NATS_USE_JETSTREAM: Enable JetStream mode for message persistence and reliability (default: false)
|
||||
export NATS_USE_JETSTREAM="true"
|
||||
```
|
||||
|
||||
### Docker Compose Configuration
|
||||
|
||||
NATS service configuration in `docker-compose.yml`:
|
||||
|
||||
```yaml
|
||||
nats:
|
||||
image: nats:2.10.24-alpine
|
||||
container_name: nats
|
||||
restart: unless-stopped
|
||||
command:
|
||||
- "--jetstream" # Enable JetStream
|
||||
- "--store_dir=/data" # Data storage directory
|
||||
- "--max_memory_store=1GB" # Memory storage limit
|
||||
- "--max_file_store=10GB" # File storage limit
|
||||
ports:
|
||||
- "4222:4222" # Client connection port
|
||||
- "8222:8222" # HTTP monitoring port
|
||||
- "6222:6222" # Cluster communication port
|
||||
volumes:
|
||||
- ./volumes/nats:/data
|
||||
networks:
|
||||
- coze-network
|
||||
healthcheck:
|
||||
test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:8222/"]
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 3
|
||||
start_period: 40s
|
||||
```
|
||||
|
||||
### Application Configuration
|
||||
|
||||
Configure NATS in Coze Studio application through environment variables:
|
||||
|
||||
```go
|
||||
// Read configuration from environment variables
|
||||
mqType := os.Getenv("COZE_MQ_TYPE")
|
||||
natsURL := os.Getenv("NATS_SERVER_URL")
|
||||
jwtToken := os.Getenv("NATS_JWT_TOKEN")
|
||||
seedFile := os.Getenv("NATS_NKEY_SEED")
|
||||
streamReplicas := os.Getenv("NATS_STREAM_REPLICAS")
|
||||
|
||||
// Create NATS EventBus
|
||||
if mqType == "nats" {
|
||||
config := &nats.Config{
|
||||
ServerURL: natsURL,
|
||||
JWTToken: jwtToken,
|
||||
SeedFile: seedFile,
|
||||
StreamReplicas: streamReplicas,
|
||||
}
|
||||
|
||||
eventBus, err := nats.NewProducer(config)
|
||||
if err != nil {
|
||||
log.Fatal("Failed to create NATS producer:", err)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Deployment Guide
|
||||
|
||||
### Docker Deployment
|
||||
|
||||
1. **Configure Environment Variables**:
|
||||
```bash
|
||||
cp docker/.env.example docker/.env
|
||||
# Edit .env file, set COZE_MQ_TYPE="nats"
|
||||
```
|
||||
|
||||
2. **Start Services**:
|
||||
```bash
|
||||
cd docker
|
||||
docker-compose up -d nats
|
||||
```
|
||||
|
||||
3. **Verify Deployment**:
|
||||
```bash
|
||||
# Check NATS service status
|
||||
docker-compose ps nats
|
||||
|
||||
# View NATS monitoring interface
|
||||
curl http://localhost:8222/varz
|
||||
```
|
||||
|
||||
### Kubernetes Deployment
|
||||
|
||||
Deploy NATS using the official Helm Chart:
|
||||
|
||||
```bash
|
||||
# Add NATS Helm repository
|
||||
helm repo add nats https://nats-io.github.io/k8s/helm/charts/
|
||||
|
||||
# Install NATS
|
||||
helm install nats nats/nats --set nats.jetstream.enabled=true
|
||||
```
|
||||
|
||||
### Production Environment Configuration
|
||||
|
||||
For production environments, the following configuration optimizations are recommended:
|
||||
|
||||
1. **Cluster Deployment**:
|
||||
```yaml
|
||||
nats:
|
||||
cluster:
|
||||
enabled: true
|
||||
replicas: 3
|
||||
```
|
||||
|
||||
2. **Persistent Storage**:
|
||||
```yaml
|
||||
nats:
|
||||
jetstream:
|
||||
fileStore:
|
||||
pvc:
|
||||
size: 100Gi
|
||||
storageClassName: fast-ssd
|
||||
```
|
||||
|
||||
3. **Resource Limits**:
|
||||
```yaml
|
||||
nats:
|
||||
resources:
|
||||
limits:
|
||||
cpu: 2000m
|
||||
memory: 4Gi
|
||||
requests:
|
||||
cpu: 500m
|
||||
memory: 1Gi
|
||||
```
|
||||
|
||||
4. **Security Configuration**:
|
||||
```yaml
|
||||
nats:
|
||||
auth:
|
||||
enabled: true
|
||||
token: "your-secure-token"
|
||||
tls:
|
||||
enabled: true
|
||||
```
|
||||
|
||||
## Monitoring and Operations
|
||||
|
||||
### Monitoring Metrics
|
||||
|
||||
NATS provides rich monitoring metrics accessible through HTTP endpoints:
|
||||
|
||||
- **Server Information**: `GET /varz`
|
||||
- **Connection Information**: `GET /connz`
|
||||
- **Subscription Information**: `GET /subsz`
|
||||
- **JetStream Information**: `GET /jsz`
|
||||
|
||||
### Key Monitoring Metrics
|
||||
|
||||
1. **Performance Metrics**:
|
||||
- Message throughput (messages/sec)
|
||||
- Message latency (latency)
|
||||
- Connection count (connections)
|
||||
|
||||
2. **Resource Metrics**:
|
||||
- Memory usage (memory usage)
|
||||
- CPU utilization (cpu usage)
|
||||
- Disk usage (disk usage)
|
||||
|
||||
3. **JetStream Metrics**:
|
||||
- Stream count (streams)
|
||||
- Consumer count (consumers)
|
||||
- Storage usage (storage usage)
|
||||
|
||||
### Log Management
|
||||
|
||||
NATS supports multiple log levels and output formats:
|
||||
|
||||
```bash
|
||||
# Enable debug logging
|
||||
nats-server --debug
|
||||
|
||||
# Log output to file
|
||||
nats-server --log /var/log/nats.log
|
||||
|
||||
# JSON format logging
|
||||
nats-server --logtime --log_size_limit 100MB
|
||||
```
|
||||
|
||||
## Performance Optimization
|
||||
|
||||
### Connection Pool Optimization
|
||||
|
||||
```go
|
||||
// Configure connection options
|
||||
opts := []nats.Option{
|
||||
nats.MaxReconnects(10),
|
||||
nats.ReconnectWait(2 * time.Second),
|
||||
nats.Timeout(5 * time.Second),
|
||||
}
|
||||
|
||||
nc, err := nats.Connect(serverURL, opts...)
|
||||
```
|
||||
|
||||
### JetStream Optimization
|
||||
|
||||
```go
|
||||
// Configure JetStream options
|
||||
jsOpts := []nats.JSOpt{
|
||||
nats.PublishAsyncMaxPending(1000),
|
||||
nats.PublishAsyncErrHandler(func(js nats.JetStream, originalMsg *nats.Msg, err error) {
|
||||
log.Printf("Async publish error: %v", err)
|
||||
}),
|
||||
}
|
||||
|
||||
js, err := nc.JetStream(jsOpts...)
|
||||
```
|
||||
|
||||
### Consumer Optimization
|
||||
|
||||
```go
|
||||
// Configure consumer options
|
||||
consumerOpts := []nats.SubOpt{
|
||||
nats.Durable("coze-consumer"),
|
||||
nats.MaxDeliver(3),
|
||||
nats.AckWait(30 * time.Second),
|
||||
nats.MaxAckPending(100),
|
||||
}
|
||||
|
||||
sub, err := js.PullSubscribe(topic, "coze-group", consumerOpts...)
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
1. **Connection Failures**:
|
||||
- Check if NATS service is running
|
||||
- Verify network connectivity
|
||||
- Confirm port configuration is correct
|
||||
|
||||
2. **Message Loss**:
|
||||
- Check if JetStream is enabled
|
||||
- Verify message acknowledgment mechanism
|
||||
- Review error logs
|
||||
|
||||
3. **Performance Issues**:
|
||||
- Monitor resource usage
|
||||
- Check for message backlog
|
||||
- Optimize consumer configuration
|
||||
|
||||
### Debugging Tools
|
||||
|
||||
NATS provides rich debugging tools:
|
||||
|
||||
```bash
|
||||
# NATS CLI tools
|
||||
nats server info
|
||||
nats stream list
|
||||
nats consumer list
|
||||
|
||||
# Monitor message flow
|
||||
nats sub "coze.>"
|
||||
nats pub "coze.test" "hello world"
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
### Subject Naming Conventions
|
||||
|
||||
Recommend using hierarchical subject naming:
|
||||
|
||||
```
|
||||
coze.workflow.{workflow_id}.{event_type}
|
||||
coze.agent.{agent_id}.{action}
|
||||
coze.knowledge.{kb_id}.{operation}
|
||||
```
|
||||
|
||||
### Error Handling
|
||||
|
||||
Implement comprehensive error handling mechanisms:
|
||||
|
||||
```go
|
||||
func (c *Consumer) handleMessage(msg *nats.Msg) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("Message processing panic: %v", r)
|
||||
msg.Nak() // Reject message, trigger retry
|
||||
}
|
||||
}()
|
||||
|
||||
if err := c.processMessage(msg.Data); err != nil {
|
||||
log.Printf("Message processing error: %v", err)
|
||||
msg.Nak()
|
||||
return
|
||||
}
|
||||
|
||||
msg.Ack() // Acknowledge successful message processing
|
||||
}
|
||||
```
|
||||
|
||||
### Resource Management
|
||||
|
||||
Properly manage NATS connections and resources:
|
||||
|
||||
```go
|
||||
func (p *Producer) Close() error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
if p.closed {
|
||||
return nil
|
||||
}
|
||||
|
||||
p.closed = true
|
||||
|
||||
if p.nc != nil {
|
||||
p.nc.Close()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
## Summary
|
||||
|
||||
NATS as Coze Studio's EventBus solution provides lightweight, high-performance, and easy-to-deploy messaging capabilities. Through JetStream extensions, NATS can also provide enterprise-grade message persistence and stream processing functionality.
|
||||
|
||||
Key advantages of choosing NATS:
|
||||
- **Simplicity**: Low deployment and maintenance costs
|
||||
- **Performance**: Extremely high message processing performance
|
||||
- **Cloud Native**: Perfect fit for containerized and Kubernetes environments
|
||||
- **Reliability**: JetStream provides message persistence and acknowledgment mechanisms
|
||||
- **Scalability**: Supports cluster deployment and horizontal scaling
|
||||
|
||||
NATS is particularly suitable for the following scenarios:
|
||||
- Inter-service communication in microservice architectures
|
||||
- Real-time data stream processing
|
||||
- Message delivery for cloud-native applications
|
||||
- Low-latency messaging systems
|
||||
- Resource-constrained deployment environments
|
||||
@@ -0,0 +1,507 @@
|
||||
# NATS EventBus 集成指南
|
||||
|
||||
## 概述
|
||||
|
||||
本文档详细介绍了 NATS 作为 EventBus 在 Coze Studio 中的集成适配情况，包括架构设计、实现细节、配置说明和使用指南。
|
||||
|
||||
## 集成背景
|
||||
|
||||
### 为什么选择 NATS？
|
||||
|
||||
在 Coze Studio 的架构中，EventBus 承担着关键的异步消息传递任务，包括工作流执行、Agent 通信、数据处理管道等核心功能。NATS 作为一个轻量级、高性能的消息系统，为 Coze Studio 带来了以下核心优势：
|
||||
|
||||
1. **轻量级**: NATS 具有极小的资源占用和简单的部署架构，非常适合云原生环境
|
||||
2. **高性能**: 提供低延迟、高吞吐量的消息传递，能够支撑 Coze Studio 大规模并发的 Agent 执行
|
||||
3. **简单易用**: API 简洁直观，降低了开发和维护成本
|
||||
4. **JetStream 支持**: 通过 JetStream 提供消息持久化、重放和流处理能力
|
||||
5. **云原生**: 原生支持 Kubernetes，易于在容器化环境中部署和管理
|
||||
6. **安全性**: 内置多种认证和授权机制，支持 TLS 加密
|
||||
|
||||
### 与其他 MQ 的对比
|
||||
|
||||
| 特性 | NATS | NSQ | Kafka | RocketMQ | Pulsar |
|
||||
| ---------------------- | -------------- | -------------- | -------------- | -------------- | -------------- |
|
||||
| **部署复杂度** | 极低 | 低 | 中等 | 中等 | 中等 |
|
||||
| **性能** | 极高 | 中等 | 高 | 高 | 高 |
|
||||
| **资源占用** | 极低 | 低 | 中等 | 中等 | 中等 |
|
||||
| **消息持久化** | JetStream | 有限 | 强 | 强 | 强 |
|
||||
| **顺序性保障** | 支持 | 弱 | 强 | 强 | 强 |
|
||||
| **水平扩展性** | 良好 | 中等 | 良好 | 良好 | 优秀 |
|
||||
| **运维复杂度** | 极低 | 低 | 高 | 中等 | 中等 |
|
||||
| **云原生支持** | 优秀 | 中等 | 中等 | 中等 | 良好 |
|
||||
|
||||
#### NATS 的核心优势
|
||||
|
||||
**轻量级和高性能**：
|
||||
- **内存占用**：NATS 服务器通常只需要几十 MB 内存即可处理数百万消息
|
||||
- **启动速度**：秒级启动，非常适合微服务和容器化部署
|
||||
- **延迟**：亚毫秒级消息延迟，适合实时性要求高的场景
|
||||
- **吞吐量**：单节点可处理数百万消息/秒
|
||||
|
||||
**简单性**：
|
||||
- **配置简单**：最小化配置即可运行，无需复杂的集群配置
|
||||
- **API 简洁**：发布/订阅模式简单直观，学习成本低
|
||||
- **运维友好**：监控和调试工具丰富，问题排查容易
|
||||
|
||||
**云原生特性**：
|
||||
- **Kubernetes 集成**：官方提供 Helm Charts 和 Operator
|
||||
- **服务发现**：内置服务发现机制，无需外部依赖
|
||||
- **弹性伸缩**：支持动态集群成员变更
|
||||
|
||||
## 架构设计
|
||||
|
||||
### 整体架构
|
||||
|
||||
```
|
||||
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
|
||||
│ Coze Studio │ │ NATS Server │ │ JetStream │
|
||||
│ Application │ │ │ │ Storage │
|
||||
├─────────────────┤ ├─────────────────┤ ├─────────────────┤
|
||||
│ Producer │───▶│ Core NATS │ │ Streams │
|
||||
│ Consumer │◀───│ JetStream │◀───│ Consumers │
|
||||
│ EventBus │ │ Clustering │ │ Key-Value │
|
||||
└─────────────────┘ └─────────────────┘ └─────────────────┘
|
||||
```
|
||||
|
||||
### 消息流转模式
|
||||
|
||||
NATS 在 Coze Studio 中支持两种消息模式：
|
||||
|
||||
1. **Core NATS**: 用于实时、轻量级的消息传递
|
||||
- 发布/订阅模式
|
||||
- 请求/响应模式
|
||||
- 队列组模式
|
||||
|
||||
2. **JetStream**: 用于需要持久化和高可靠性的消息
|
||||
- 流式存储
|
||||
- 消息重放
|
||||
- 消费者确认机制
|
||||
|
||||
## 实现细节
|
||||
|
||||
### Producer 实现
|
||||
|
||||
Producer 负责向 NATS 发送消息，支持以下特性：
|
||||
|
||||
```go
|
||||
type Producer struct {
|
||||
nc nats.Conn
|
||||
js nats.JetStreamContext
|
||||
closed bool
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
func (p *Producer) SendMessage(ctx context.Context, topic string, message []byte) error {
|
||||
// 支持 Core NATS 和 JetStream 两种模式
|
||||
if p.js != nil {
|
||||
// JetStream 模式：支持消息持久化
|
||||
_, err := p.js.Publish(topic, message)
|
||||
return err
|
||||
} else {
|
||||
// Core NATS 模式：轻量级发布
|
||||
return p.nc.Publish(topic, message)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Consumer 实现
|
||||
|
||||
Consumer 负责从 NATS 接收和处理消息：
|
||||
|
||||
```go
|
||||
func (c *Consumer) RegisterConsumer(serverURL, topic, group string, handler ConsumerHandler) error {
|
||||
// 根据配置选择 JetStream 或 Core NATS
|
||||
if c.useJetStream {
|
||||
return c.startJetStreamConsumer(ctx, topic, group, handler)
|
||||
} else {
|
||||
return c.startCoreConsumer(ctx, topic, group, handler)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### JetStream Consumer 特性
|
||||
|
||||
- **消息确认**: 支持手动确认机制，确保消息处理成功
|
||||
- **重试机制**: 失败消息自动重试，支持指数退避
|
||||
- **顺序处理**: 单条消息处理，避免批处理带来的复杂性
|
||||
- **流控制**: 精确的消息流控制，防止消费者过载
|
||||
|
||||
#### Core NATS Consumer 特性
|
||||
|
||||
- **队列组**: 支持负载均衡的消息分发
|
||||
- **轻量级**: 无持久化开销，适合实时消息处理
|
||||
- **高性能**: 极低的消息处理延迟
|
||||
|
||||
## 配置说明
|
||||
|
||||
### 环境变量配置
|
||||
|
||||
在 `docker/.env.example` 中添加以下 NATS 相关配置：
|
||||
|
||||
```bash
|
||||
# Backend Event Bus
|
||||
export COZE_MQ_TYPE="nats" # 设置消息队列类型为 NATS
|
||||
export MQ_NAME_SERVER="nats:4222" # NATS 服务器地址
|
||||
|
||||
# NATS 特定配置
|
||||
# NATS_SERVER_URL: NATS 服务器连接 URL，支持 nats:// 和 tls:// 协议
|
||||
# 集群模式使用逗号分隔的 URL: "nats://nats1:4222,nats://nats2:4222"
|
||||
# TLS 连接: "tls://nats:4222"
|
||||
export NATS_SERVER_URL="nats://nats:4222"
|
||||
|
||||
# NATS_JWT_TOKEN: NATS JWT 认证令牌（留空表示无认证）
|
||||
export NATS_JWT_TOKEN=""
|
||||
|
||||
# NATS_NKEY_SEED: NATS NKey 认证种子文件路径（可选）
|
||||
export NATS_NKEY_SEED=""
|
||||
|
||||
# NATS_USERNAME: NATS 用户名认证（可选）
|
||||
export NATS_USERNAME=""
|
||||
|
||||
# NATS_PASSWORD: NATS 密码认证（可选）
|
||||
export NATS_PASSWORD=""
|
||||
|
||||
# NATS_TOKEN: NATS 令牌认证（可选）
|
||||
export NATS_TOKEN=""
|
||||
|
||||
# NATS_STREAM_REPLICAS: JetStream 流的副本数量（默认: 1）
|
||||
export NATS_STREAM_REPLICAS="1"
|
||||
|
||||
# NATS_USE_JETSTREAM: 启用 JetStream 模式以获得消息持久化和可靠性（默认: false）
|
||||
export NATS_USE_JETSTREAM="true"
|
||||
```
|
||||
|
||||
### Docker Compose 配置
|
||||
|
||||
在 `docker-compose.yml` 中的 NATS 服务配置：
|
||||
|
||||
```yaml
|
||||
nats:
|
||||
image: nats:2.10.24-alpine
|
||||
container_name: nats
|
||||
restart: unless-stopped
|
||||
command:
|
||||
- "--jetstream" # 启用 JetStream
|
||||
- "--store_dir=/data" # 数据存储目录
|
||||
- "--max_memory_store=1GB" # 内存存储限制
|
||||
- "--max_file_store=10GB" # 文件存储限制
|
||||
ports:
|
||||
- "4222:4222" # 客户端连接端口
|
||||
- "8222:8222" # HTTP 监控端口
|
||||
- "6222:6222" # 集群通信端口
|
||||
volumes:
|
||||
- ./volumes/nats:/data
|
||||
networks:
|
||||
- coze-network
|
||||
healthcheck:
|
||||
test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:8222/"]
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 3
|
||||
start_period: 40s
|
||||
```
|
||||
|
||||
### 应用程序配置
|
||||
|
||||
在 Coze Studio 应用中，通过环境变量配置 NATS：
|
||||
|
||||
```go
|
||||
// 从环境变量读取配置
|
||||
mqType := os.Getenv("COZE_MQ_TYPE")
|
||||
natsURL := os.Getenv("NATS_SERVER_URL")
|
||||
jwtToken := os.Getenv("NATS_JWT_TOKEN")
|
||||
seedFile := os.Getenv("NATS_NKEY_SEED")
|
||||
streamReplicas := os.Getenv("NATS_STREAM_REPLICAS")
|
||||
|
||||
// 创建 NATS EventBus
|
||||
if mqType == "nats" {
|
||||
config := &nats.Config{
|
||||
ServerURL: natsURL,
|
||||
JWTToken: jwtToken,
|
||||
SeedFile: seedFile,
|
||||
StreamReplicas: streamReplicas,
|
||||
}
|
||||
|
||||
eventBus, err := nats.NewProducer(config)
|
||||
if err != nil {
|
||||
log.Fatal("Failed to create NATS producer:", err)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 部署指南
|
||||
|
||||
### Docker 部署
|
||||
|
||||
1. **配置环境变量**：
|
||||
```bash
|
||||
cp docker/.env.example docker/.env
|
||||
# 编辑 .env 文件，设置 COZE_MQ_TYPE="nats"
|
||||
```
|
||||
|
||||
2. **启动服务**：
|
||||
```bash
|
||||
cd docker
|
||||
docker-compose up -d nats
|
||||
```
|
||||
|
||||
3. **验证部署**：
|
||||
```bash
|
||||
# 检查 NATS 服务状态
|
||||
docker-compose ps nats
|
||||
|
||||
# 查看 NATS 监控界面
|
||||
curl http://localhost:8222/varz
|
||||
```
|
||||
|
||||
### Kubernetes 部署
|
||||
|
||||
使用官方 Helm Chart 部署 NATS：
|
||||
|
||||
```bash
|
||||
# 添加 NATS Helm 仓库
|
||||
helm repo add nats https://nats-io.github.io/k8s/helm/charts/
|
||||
|
||||
# 安装 NATS
|
||||
helm install nats nats/nats --set nats.jetstream.enabled=true
|
||||
```
|
||||
|
||||
### 生产环境配置
|
||||
|
||||
对于生产环境，建议进行以下配置优化：
|
||||
|
||||
1. **集群部署**：
|
||||
```yaml
|
||||
nats:
|
||||
cluster:
|
||||
enabled: true
|
||||
replicas: 3
|
||||
```
|
||||
|
||||
2. **持久化存储**：
|
||||
```yaml
|
||||
nats:
|
||||
jetstream:
|
||||
fileStore:
|
||||
pvc:
|
||||
size: 100Gi
|
||||
storageClassName: fast-ssd
|
||||
```
|
||||
|
||||
3. **资源限制**：
|
||||
```yaml
|
||||
nats:
|
||||
resources:
|
||||
limits:
|
||||
cpu: 2000m
|
||||
memory: 4Gi
|
||||
requests:
|
||||
cpu: 500m
|
||||
memory: 1Gi
|
||||
```
|
||||
|
||||
4. **安全配置**：
|
||||
```yaml
|
||||
nats:
|
||||
auth:
|
||||
enabled: true
|
||||
token: "your-secure-token"
|
||||
tls:
|
||||
enabled: true
|
||||
```
|
||||
|
||||
## 监控和运维
|
||||
|
||||
### 监控指标
|
||||
|
||||
NATS 提供丰富的监控指标，可通过 HTTP 端点获取：
|
||||
|
||||
- **服务器信息**: `GET /varz`
|
||||
- **连接信息**: `GET /connz`
|
||||
- **订阅信息**: `GET /subsz`
|
||||
- **JetStream 信息**: `GET /jsz`
|
||||
|
||||
### 关键监控指标
|
||||
|
||||
1. **性能指标**：
|
||||
- 消息吞吐量 (messages/sec)
|
||||
- 消息延迟 (latency)
|
||||
- 连接数 (connections)
|
||||
|
||||
2. **资源指标**：
|
||||
- 内存使用量 (memory usage)
|
||||
- CPU 使用率 (cpu usage)
|
||||
- 磁盘使用量 (disk usage)
|
||||
|
||||
3. **JetStream 指标**：
|
||||
- 流数量 (streams)
|
||||
- 消费者数量 (consumers)
|
||||
- 存储使用量 (storage usage)
|
||||
|
||||
### 日志管理
|
||||
|
||||
NATS 支持多种日志级别和输出格式：
|
||||
|
||||
```bash
|
||||
# 启用调试日志
|
||||
nats-server --debug
|
||||
|
||||
# 日志输出到文件
|
||||
nats-server --log /var/log/nats.log
|
||||
|
||||
# JSON 格式日志
|
||||
nats-server --logtime --log_size_limit 100MB
|
||||
```
|
||||
|
||||
## 性能优化
|
||||
|
||||
### 连接池优化
|
||||
|
||||
```go
|
||||
// 配置连接选项
|
||||
opts := []nats.Option{
|
||||
nats.MaxReconnects(10),
|
||||
nats.ReconnectWait(2 * time.Second),
|
||||
nats.Timeout(5 * time.Second),
|
||||
}
|
||||
|
||||
nc, err := nats.Connect(serverURL, opts...)
|
||||
```
|
||||
|
||||
### JetStream 优化
|
||||
|
||||
```go
|
||||
// 配置 JetStream 选项
|
||||
jsOpts := []nats.JSOpt{
|
||||
nats.PublishAsyncMaxPending(1000),
|
||||
nats.PublishAsyncErrHandler(func(js nats.JetStream, originalMsg *nats.Msg, err error) {
|
||||
log.Printf("Async publish error: %v", err)
|
||||
}),
|
||||
}
|
||||
|
||||
js, err := nc.JetStream(jsOpts...)
|
||||
```
|
||||
|
||||
### 消费者优化
|
||||
|
||||
```go
|
||||
// 配置消费者选项
|
||||
consumerOpts := []nats.SubOpt{
|
||||
nats.Durable("coze-consumer"),
|
||||
nats.MaxDeliver(3),
|
||||
nats.AckWait(30 * time.Second),
|
||||
nats.MaxAckPending(100),
|
||||
}
|
||||
|
||||
sub, err := js.PullSubscribe(topic, "coze-group", consumerOpts...)
|
||||
```
|
||||
|
||||
## 故障排查
|
||||
|
||||
### 常见问题
|
||||
|
||||
1. **连接失败**：
|
||||
- 检查 NATS 服务是否启动
|
||||
- 验证网络连通性
|
||||
- 确认端口配置正确
|
||||
|
||||
2. **消息丢失**：
|
||||
- 检查 JetStream 是否启用
|
||||
- 验证消息确认机制
|
||||
- 查看错误日志
|
||||
|
||||
3. **性能问题**：
|
||||
- 监控资源使用情况
|
||||
- 检查消息积压
|
||||
- 优化消费者配置
|
||||
|
||||
### 调试工具
|
||||
|
||||
NATS 提供了丰富的调试工具：
|
||||
|
||||
```bash
|
||||
# NATS CLI 工具
|
||||
nats server info
|
||||
nats stream list
|
||||
nats consumer list
|
||||
|
||||
# 监控消息流
|
||||
nats sub "coze.>"
|
||||
nats pub "coze.test" "hello world"
|
||||
```
|
||||
|
||||
## 最佳实践
|
||||
|
||||
### 主题命名规范
|
||||
|
||||
建议使用层次化的主题命名：
|
||||
|
||||
```
|
||||
coze.workflow.{workflow_id}.{event_type}
|
||||
coze.agent.{agent_id}.{action}
|
||||
coze.knowledge.{kb_id}.{operation}
|
||||
```
|
||||
|
||||
### 错误处理
|
||||
|
||||
实现完善的错误处理机制：
|
||||
|
||||
```go
|
||||
func (c *Consumer) handleMessage(msg *nats.Msg) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("Message processing panic: %v", r)
|
||||
msg.Nak() // 拒绝消息，触发重试
|
||||
}
|
||||
}()
|
||||
|
||||
if err := c.processMessage(msg.Data); err != nil {
|
||||
log.Printf("Message processing error: %v", err)
|
||||
msg.Nak()
|
||||
return
|
||||
}
|
||||
|
||||
msg.Ack() // 确认消息处理成功
|
||||
}
|
||||
```
|
||||
|
||||
### 资源管理
|
||||
|
||||
正确管理 NATS 连接和资源：
|
||||
|
||||
```go
|
||||
func (p *Producer) Close() error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
if p.closed {
|
||||
return nil
|
||||
}
|
||||
|
||||
p.closed = true
|
||||
|
||||
if p.nc != nil {
|
||||
p.nc.Close()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
## 总结
|
||||
|
||||
NATS 作为 Coze Studio 的 EventBus 解决方案，提供了轻量级、高性能、易于部署的消息传递能力。通过 JetStream 扩展，NATS 还能提供企业级的消息持久化和流处理功能。
|
||||
|
||||
选择 NATS 的主要优势：
|
||||
- **简单性**: 部署和维护成本低
|
||||
- **性能**: 极高的消息处理性能
|
||||
- **云原生**: 完美适配容器化和 Kubernetes 环境
|
||||
- **可靠性**: JetStream 提供消息持久化和确认机制
|
||||
- **扩展性**: 支持集群部署和水平扩展
|
||||
|
||||
NATS 特别适合以下场景：
|
||||
- 微服务架构的服务间通信
|
||||
- 实时数据流处理
|
||||
- 云原生应用的消息传递
|
||||
- 需要低延迟的消息系统
|
||||
- 资源受限的部署环境
|
||||
@@ -0,0 +1,622 @@
|
||||
# OceanBase Vector Database Integration Guide
|
||||
|
||||
## Overview
|
||||
|
||||
This document provides a comprehensive guide to the integration of OceanBase vector database in Coze Studio, including architectural design, implementation details, configuration instructions, and usage guidelines.
|
||||
|
||||
## Integration Background
|
||||
|
||||
### Why Choose OceanBase?
|
||||
|
||||
1. **Transaction Support**: OceanBase provides complete ACID transaction support, ensuring data consistency
|
||||
2. **Simple Deployment**: Compared to specialized vector databases like Milvus, OceanBase deployment is simpler
|
||||
3. **MySQL Compatibility**: Compatible with MySQL protocol, low learning curve
|
||||
4. **Vector Extensions**: Native support for vector data types and indexing
|
||||
5. **Operations Friendly**: Low operational costs, suitable for small to medium-scale applications
|
||||
|
||||
### Comparison with Milvus
|
||||
|
||||
| Feature | OceanBase | Milvus |
|
||||
| ------------------------------- | -------------------- | --------------------------- |
|
||||
| **Deployment Complexity** | Low (Single Machine) | High (Requires etcd, MinIO) |
|
||||
| **Transaction Support** | Full ACID | Limited |
|
||||
| **Vector Search Speed** | Medium | Faster |
|
||||
| **Storage Efficiency** | Medium | Higher |
|
||||
| **Operational Cost** | Low | High |
|
||||
| **Learning Curve** | Gentle | Steep |
|
||||
|
||||
## Architectural Design
|
||||
|
||||
### Overall Architecture
|
||||
|
||||
```
|
||||
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
|
||||
│ Coze Studio │ │ OceanBase │ │ Vector Store │
|
||||
│ Application │───▶│ Client │───▶│ Manager │
|
||||
└─────────────────┘ └─────────────────┘ └─────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────┐
|
||||
│ OceanBase │
|
||||
│ Database │
|
||||
└─────────────────┘
|
||||
```
|
||||
|
||||
### Core Components
|
||||
|
||||
#### 1. OceanBase Client (`backend/infra/impl/oceanbase/`)
|
||||
|
||||
**Main Files**:
|
||||
|
||||
- `oceanbase.go` - Delegation client, providing backward-compatible interface
|
||||
- `oceanbase_official.go` - Core implementation, based on official documentation
|
||||
- `types.go` - Type definitions
|
||||
|
||||
**Core Functions**:
|
||||
|
||||
```go
|
||||
type OceanBaseClient interface {
|
||||
CreateCollection(ctx context.Context, collectionName string) error
|
||||
InsertVectors(ctx context.Context, collectionName string, vectors []VectorResult) error
|
||||
SearchVectors(ctx context.Context, collectionName string, queryVector []float64, topK int) ([]VectorResult, error)
|
||||
DeleteVector(ctx context.Context, collectionName string, vectorID string) error
|
||||
InitDatabase(ctx context.Context) error
|
||||
DropCollection(ctx context.Context, collectionName string) error
|
||||
}
|
||||
```
|
||||
|
||||
#### 2. Search Store Manager (`backend/infra/impl/document/searchstore/oceanbase/`)
|
||||
|
||||
**Main Files**:
|
||||
|
||||
- `oceanbase_manager.go` - Manager implementation
|
||||
- `oceanbase_searchstore.go` - Search store implementation
|
||||
- `factory.go` - Factory pattern creation
|
||||
- `consts.go` - Constant definitions
|
||||
- `convert.go` - Data conversion
|
||||
- `register.go` - Registration functions
|
||||
|
||||
**Core Functions**:
|
||||
|
||||
```go
|
||||
type Manager interface {
|
||||
Create(ctx context.Context, collectionName string) (SearchStore, error)
|
||||
Get(ctx context.Context, collectionName string) (SearchStore, error)
|
||||
Delete(ctx context.Context, collectionName string) error
|
||||
}
|
||||
```
|
||||
|
||||
#### 3. Application Layer Integration (`backend/application/base/appinfra/`)
|
||||
|
||||
**File**: `app_infra.go`
|
||||
|
||||
**Integration Point**:
|
||||
|
||||
```go
|
||||
case "oceanbase":
|
||||
// Build DSN
|
||||
dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local",
|
||||
user, password, host, port, database)
|
||||
|
||||
// Create client
|
||||
client, err := oceanbaseClient.NewOceanBaseClient(dsn)
|
||||
|
||||
// Initialize database
|
||||
if err := client.InitDatabase(ctx); err != nil {
|
||||
return nil, fmt.Errorf("init oceanbase database failed, err=%w", err)
|
||||
}
|
||||
```
|
||||
|
||||
## Configuration Instructions
|
||||
|
||||
### Environment Variable Configuration
|
||||
|
||||
#### Required Configuration
|
||||
|
||||
```bash
|
||||
# Vector store type
|
||||
VECTOR_STORE_TYPE=oceanbase
|
||||
|
||||
# OceanBase connection configuration
|
||||
OCEANBASE_HOST=localhost
|
||||
OCEANBASE_PORT=2881
|
||||
OCEANBASE_USER=root
|
||||
OCEANBASE_PASSWORD=coze123
|
||||
OCEANBASE_DATABASE=test
|
||||
```
|
||||
|
||||
#### Optional Configuration
|
||||
|
||||
```bash
|
||||
# Performance optimization configuration
|
||||
OCEANBASE_VECTOR_MEMORY_LIMIT_PERCENTAGE=30
|
||||
OCEANBASE_BATCH_SIZE=100
|
||||
OCEANBASE_MAX_OPEN_CONNS=100
|
||||
OCEANBASE_MAX_IDLE_CONNS=10
|
||||
|
||||
# Cache configuration
|
||||
OCEANBASE_ENABLE_CACHE=true
|
||||
OCEANBASE_CACHE_TTL=300
|
||||
|
||||
# Monitoring configuration
|
||||
OCEANBASE_ENABLE_METRICS=true
|
||||
OCEANBASE_ENABLE_SLOW_QUERY_LOG=true
|
||||
|
||||
# Retry configuration
|
||||
OCEANBASE_MAX_RETRIES=3
|
||||
OCEANBASE_RETRY_DELAY=1
|
||||
OCEANBASE_CONN_TIMEOUT=30
|
||||
```
|
||||
|
||||
### Docker Configuration
|
||||
|
||||
#### docker-compose-oceanbase.yml
|
||||
|
||||
```yaml
|
||||
oceanbase:
|
||||
image: oceanbase/oceanbase-ce:latest
|
||||
container_name: coze-oceanbase
|
||||
environment:
|
||||
MODE: SLIM
|
||||
OB_DATAFILE_SIZE: 1G
|
||||
OB_SYS_PASSWORD: ${OCEANBASE_PASSWORD:-coze123}
|
||||
OB_TENANT_PASSWORD: ${OCEANBASE_PASSWORD:-coze123}
|
||||
ports:
|
||||
- '2881:2881'
|
||||
volumes:
|
||||
- ./data/oceanbase/ob:/root/ob
|
||||
- ./data/oceanbase/cluster:/root/.obd/cluster
|
||||
deploy:
|
||||
resources:
|
||||
limits:
|
||||
memory: 4G
|
||||
reservations:
|
||||
memory: 2G
|
||||
```
|
||||
|
||||
## Usage Guide
|
||||
|
||||
### 1. Quick Start
|
||||
|
||||
```bash
|
||||
# Clone the project
|
||||
git clone https://github.com/coze-dev/coze-studio.git
|
||||
cd coze-studio
|
||||
|
||||
# Setup OceanBase environment
|
||||
make oceanbase_env
|
||||
|
||||
# Start OceanBase debug environment
|
||||
make oceanbase_debug
|
||||
```
|
||||
|
||||
### 2. Verify Deployment
|
||||
|
||||
```bash
|
||||
# Check container status
|
||||
docker ps | grep oceanbase
|
||||
|
||||
# Test connection
|
||||
mysql -h localhost -P 2881 -u root -p -e "SELECT 1;"
|
||||
|
||||
# View databases
|
||||
mysql -h localhost -P 2881 -u root -p -e "SHOW DATABASES;"
|
||||
```
|
||||
|
||||
### 3. Create Knowledge Base
|
||||
|
||||
In the Coze Studio interface:
|
||||
|
||||
1. Enter knowledge base management
|
||||
2. Select OceanBase as vector storage
|
||||
3. Upload documents for vectorization
|
||||
4. Test vector retrieval functionality
|
||||
|
||||
### 4. Performance Monitoring
|
||||
|
||||
```bash
|
||||
# View container resource usage
|
||||
docker stats coze-oceanbase
|
||||
|
||||
# View slow query logs
|
||||
docker logs coze-oceanbase | grep "slow query"
|
||||
|
||||
# View connection count
|
||||
mysql -h localhost -P 2881 -u root -p -e "SHOW PROCESSLIST;"
|
||||
```
|
||||
|
||||
## Helm Deployment Guide (Kubernetes)
|
||||
|
||||
### 1. Environment Preparation
|
||||
|
||||
Ensure the following tools are installed:
|
||||
|
||||
- Kubernetes cluster (recommended: k3s or kind)
|
||||
- Helm 3.x
|
||||
- kubectl
|
||||
|
||||
### 2. Install Dependencies
|
||||
|
||||
#### Install cert-manager
|
||||
|
||||
```bash
|
||||
# Add cert-manager Helm repository
|
||||
helm repo add jetstack https://charts.jetstack.io
|
||||
helm repo update
|
||||
|
||||
# Install cert-manager
|
||||
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.16.2/cert-manager.yaml
|
||||
|
||||
# Wait for cert-manager to be ready
|
||||
kubectl wait --for=condition=ready pod -l app.kubernetes.io/name=cert-manager -n cert-manager --timeout=300s
|
||||
```
|
||||
|
||||
#### Install ob-operator
|
||||
|
||||
```bash
|
||||
# Add ob-operator Helm repository
|
||||
helm repo add ob-operator https://oceanbase.github.io/ob-operator/
|
||||
helm repo update
|
||||
|
||||
# Install ob-operator
|
||||
helm install ob-operator ob-operator/ob-operator --set reporter=cozeAi --namespace=oceanbase-system --create-namespace
|
||||
|
||||
# Wait for ob-operator to be ready
|
||||
kubectl wait --for=condition=ready pod -l control-plane=controller-manager -n oceanbase-system --timeout=300s
|
||||
```
|
||||
|
||||
### 3. Deploy OceanBase
|
||||
|
||||
#### Using Integrated Helm Chart
|
||||
|
||||
```bash
|
||||
# Deploy complete Coze Studio application (including OceanBase)
|
||||
helm install coze-studio helm/charts/opencoze \
|
||||
--set oceanbase.enabled=true \
|
||||
--namespace coze-studio \
|
||||
--create-namespace
|
||||
|
||||
# Or deploy only OceanBase component
|
||||
helm install oceanbase-only helm/charts/opencoze \
|
||||
--set oceanbase.enabled=true \
|
||||
--set mysql.enabled=false \
|
||||
--set redis.enabled=false \
|
||||
--set minio.enabled=false \
|
||||
--set elasticsearch.enabled=false \
|
||||
--set milvus.enabled=false \
|
||||
--set rocketmq.enabled=false \
|
||||
--namespace oceanbase \
|
||||
--create-namespace
|
||||
```
|
||||
|
||||
#### Custom Configuration
|
||||
|
||||
Create `oceanbase-values.yaml` file:
|
||||
|
||||
```yaml
|
||||
oceanbase:
|
||||
enabled: true
|
||||
port: 2881
|
||||
targetPort: 2881
|
||||
clusterName: 'cozeAi'
|
||||
clusterId: 1
|
||||
image:
|
||||
repository: oceanbase/oceanbase-ce
|
||||
tag: 'latest'
|
||||
obAgentVersion: '4.2.2-100000042024011120'
|
||||
monitorEnabled: true
|
||||
storageClass: ''
|
||||
observerConfig:
|
||||
resource:
|
||||
cpu: 2
|
||||
memory: 8Gi
|
||||
storages:
|
||||
dataStorage: 10G
|
||||
redoLogStorage: 5G
|
||||
logStorage: 5G
|
||||
monitorResource:
|
||||
cpu: 100m
|
||||
memory: 256Mi
|
||||
generateUserSecrets: true
|
||||
userSecrets:
|
||||
root: 'coze123'
|
||||
monitor: 'coze123'
|
||||
operator: 'coze123'
|
||||
proxyro: 'coze123'
|
||||
topology:
|
||||
- zone: zone1
|
||||
replica: 1
|
||||
parameters:
|
||||
- name: system_memory
|
||||
value: '4G'
|
||||
- name: '__min_full_resource_pool_memory'
|
||||
value: '4294967296'
|
||||
annotations: {}
|
||||
backupVolumeEnabled: false
|
||||
```
|
||||
|
||||
Deploy with custom configuration:
|
||||
|
||||
```bash
|
||||
helm install oceanbase-custom helm/charts/opencoze \
|
||||
-f oceanbase-values.yaml \
|
||||
--namespace oceanbase \
|
||||
--create-namespace
|
||||
```
|
||||
|
||||
### 4. Verify Deployment
|
||||
|
||||
```bash
|
||||
# Check OBCluster status
|
||||
kubectl get obcluster -n oceanbase
|
||||
|
||||
# Check OceanBase pods
|
||||
kubectl get pods -n oceanbase
|
||||
|
||||
# Check services
|
||||
kubectl get svc -n oceanbase
|
||||
|
||||
# View detailed status
|
||||
kubectl describe obcluster -n oceanbase
|
||||
```
|
||||
|
||||
### 5. Connection Testing
|
||||
|
||||
#### Port Forwarding
|
||||
|
||||
```bash
|
||||
# Forward OceanBase port
|
||||
kubectl port-forward svc/oceanbase-service -n oceanbase 2881:2881
|
||||
```
|
||||
|
||||
#### Using obclient Connection
|
||||
|
||||
```bash
|
||||
# Connect within cluster
|
||||
kubectl exec -it deployment/oceanbase-obcluster-zone1 -n oceanbase -- obclient -h127.0.0.1 -P2881 -uroot@test -pcoze123 -Dtest
|
||||
|
||||
# Connect from external (requires port forwarding)
|
||||
obclient -h127.0.0.1 -P2881 -uroot@test -pcoze123 -Dtest
|
||||
```
|
||||
|
||||
#### Using MySQL Client Connection
|
||||
|
||||
```bash
|
||||
# Using MySQL client
|
||||
mysql -h127.0.0.1 -P2881 -uroot@test -pcoze123 -Dtest
|
||||
```
|
||||
|
||||
### 6. Monitoring and Management
|
||||
|
||||
#### View Logs
|
||||
|
||||
```bash
|
||||
# View OceanBase logs
|
||||
kubectl logs -f deployment/oceanbase-obcluster-zone1 -n oceanbase
|
||||
|
||||
# View ob-operator logs
|
||||
kubectl logs -f deployment/oceanbase-controller-manager -n oceanbase-system
|
||||
```
|
||||
|
||||
#### Scaling
|
||||
|
||||
```bash
|
||||
# Scale replica count
|
||||
kubectl patch obcluster oceanbase-obcluster -n oceanbase --type='merge' -p='{"spec":{"topology":[{"zone":"zone1","replica":2}]}}'
|
||||
|
||||
# Adjust resource configuration
|
||||
kubectl patch obcluster oceanbase-obcluster -n oceanbase --type='merge' -p='{"spec":{"observer":{"resource":{"cpu":4,"memory":"16Gi"}}}}'
|
||||
```
|
||||
|
||||
#### Backup and Recovery
|
||||
|
||||
```bash
|
||||
# Create backup
|
||||
kubectl apply -f - <<EOF
|
||||
apiVersion: oceanbase.oceanbase.com/v1alpha1
|
||||
kind: OBTenantBackupPolicy
|
||||
metadata:
|
||||
name: backup-policy
|
||||
namespace: oceanbase
|
||||
spec:
|
||||
obClusterName: oceanbase-obcluster
|
||||
tenantName: test
|
||||
backupType: FULL
|
||||
schedule: "0 2 * * *"
|
||||
destination:
|
||||
path: "file:///backup"
|
||||
EOF
|
||||
```
|
||||
|
||||
### 7. Troubleshooting
|
||||
|
||||
#### Common Issues
|
||||
|
||||
1. **OBCluster Creation Failed**
|
||||
|
||||
```bash
|
||||
# Check ob-operator status
|
||||
kubectl get pods -n oceanbase-system
|
||||
|
||||
# View detailed errors
|
||||
kubectl describe obcluster -n oceanbase
|
||||
```
|
||||
2. **Image Pull Failed**
|
||||
|
||||
```bash
|
||||
# Check node image pull capability
|
||||
kubectl describe node
|
||||
|
||||
# Manually pull image
|
||||
docker pull oceanbase/oceanbase-cloud-native:4.3.5.3-103000092025080818
|
||||
```
|
||||
3. **Storage Issues**
|
||||
|
||||
```bash
|
||||
# Check PVC status
|
||||
kubectl get pvc -n oceanbase
|
||||
|
||||
# Check storage class
|
||||
kubectl get storageclass
|
||||
```
|
||||
|
||||
#### Log Analysis
|
||||
|
||||
```bash
|
||||
# View all related logs
|
||||
kubectl logs -f deployment/oceanbase-controller-manager -n oceanbase-system
|
||||
kubectl logs -f deployment/oceanbase-obcluster-zone1 -n oceanbase
|
||||
kubectl logs -f deployment/cert-manager -n cert-manager
|
||||
```
|
||||
|
||||
### 8. Uninstallation
|
||||
|
||||
```bash
|
||||
# Uninstall OceanBase
|
||||
helm uninstall oceanbase-custom -n oceanbase
|
||||
|
||||
# Delete namespace
|
||||
kubectl delete namespace oceanbase
|
||||
|
||||
# Uninstall ob-operator
|
||||
helm uninstall ob-operator -n oceanbase-system
|
||||
|
||||
# Uninstall cert-manager
|
||||
kubectl delete -f https://github.com/cert-manager/cert-manager/releases/download/v1.16.2/cert-manager.yaml
|
||||
```
|
||||
|
||||
## Integration Features
|
||||
|
||||
### 1. Design Principles
|
||||
|
||||
#### Architecture Compatibility Design
|
||||
|
||||
- Strictly follow Coze Studio core architectural design principles, ensuring seamless integration of OceanBase adaptation layer with existing systems
|
||||
- Adopt delegation pattern (Delegation Pattern) to achieve backward compatibility, ensuring stability and consistency of existing interfaces
|
||||
- Maintain complete compatibility with existing vector storage interfaces, ensuring smooth system migration and upgrade
|
||||
|
||||
#### Performance First
|
||||
|
||||
- Use HNSW index to achieve efficient approximate nearest neighbor search
|
||||
- Batch operations reduce database interaction frequency
|
||||
- Connection pool management optimizes resource usage
|
||||
|
||||
#### Easy Deployment
|
||||
|
||||
- Single machine deployment, no complex cluster configuration required
|
||||
- Docker one-click deployment
|
||||
- Environment variable configuration, flexible and easy to use
|
||||
|
||||
### 2. Technical Highlights
|
||||
|
||||
#### Delegation Pattern Design
|
||||
|
||||
```go
|
||||
type OceanBaseClient struct {
|
||||
official *OceanBaseOfficialClient
|
||||
}
|
||||
|
||||
func (c *OceanBaseClient) CreateCollection(ctx context.Context, collectionName string) error {
|
||||
return c.official.CreateCollection(ctx, collectionName)
|
||||
}
|
||||
```
|
||||
|
||||
#### Intelligent Configuration Management
|
||||
|
||||
```go
|
||||
func DefaultConfig() *Config {
|
||||
return &Config{
|
||||
Host: getEnv("OCEANBASE_HOST", "localhost"),
|
||||
Port: getEnvAsInt("OCEANBASE_PORT", 2881),
|
||||
User: getEnv("OCEANBASE_USER", "root"),
|
||||
Password: getEnv("OCEANBASE_PASSWORD", ""),
|
||||
Database: getEnv("OCEANBASE_DATABASE", "test"),
|
||||
// ... other configurations
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Error Handling Optimization
|
||||
|
||||
```go
|
||||
func (c *OceanBaseOfficialClient) setVectorParameters() error {
|
||||
params := map[string]string{
|
||||
"ob_vector_memory_limit_percentage": "30",
|
||||
"ob_query_timeout": "86400000000",
|
||||
"max_allowed_packet": "1073741824",
|
||||
}
|
||||
|
||||
for param, value := range params {
|
||||
if err := c.db.Exec(fmt.Sprintf("SET GLOBAL %s = %s", param, value)).Error; err != nil {
|
||||
log.Printf("Warning: Failed to set %s: %v", param, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### 1. Common Issues
|
||||
|
||||
#### Connection Issues
|
||||
|
||||
```bash
|
||||
# Check container status
|
||||
docker ps | grep oceanbase
|
||||
|
||||
# Check port mapping
|
||||
docker port coze-oceanbase
|
||||
|
||||
# Test connection
|
||||
mysql -h localhost -P 2881 -u root -p -e "SELECT 1;"
|
||||
```
|
||||
|
||||
#### Vector Index Issues
|
||||
|
||||
```sql
|
||||
-- Check index status
|
||||
SHOW INDEX FROM test_vectors;
|
||||
|
||||
-- Rebuild index
|
||||
DROP INDEX idx_test_embedding ON test_vectors;
|
||||
CREATE VECTOR INDEX idx_test_embedding ON test_vectors(embedding)
|
||||
WITH (distance=cosine, type=hnsw, lib=vsag, m=16, ef_construction=200, ef_search=64);
|
||||
```
|
||||
|
||||
#### Performance Issues
|
||||
|
||||
```sql
|
||||
-- Adjust memory limit
|
||||
SET GLOBAL ob_vector_memory_limit_percentage = 50;
|
||||
|
||||
-- View slow queries
|
||||
SHOW VARIABLES LIKE 'slow_query_log';
|
||||
```
|
||||
|
||||
### 2. Log Analysis
|
||||
|
||||
```bash
|
||||
# View OceanBase logs
|
||||
docker logs coze-oceanbase
|
||||
|
||||
# View application logs
|
||||
tail -f logs/coze-studio.log | grep -i "oceanbase\|vector"
|
||||
```
|
||||
|
||||
## Summary
|
||||
|
||||
The integration of OceanBase vector database in Coze Studio has achieved the following goals:
|
||||
|
||||
1. **Complete Functionality**: Supports complete vector storage and retrieval functionality
|
||||
2. **Good Performance**: Achieves efficient vector search through HNSW indexing
|
||||
3. **Simple Deployment**: Single machine deployment, no complex configuration required
|
||||
4. **Operations Friendly**: Low operational costs, easy monitoring and management
|
||||
5. **Strong Scalability**: Supports horizontal and vertical scaling
|
||||
|
||||
Through this integration, Coze Studio provides users with a simple, efficient, and reliable vector database solution, particularly suitable for scenarios requiring transaction support, simple deployment, and low operational costs.
|
||||
|
||||
## Related Links
|
||||
|
||||
- [OceanBase Official Documentation](https://www.oceanbase.com/docs)
|
||||
- [Coze Studio Project Repository](https://github.com/coze-dev/coze-studio)
|
||||
@@ -0,0 +1,622 @@
|
||||
# OceanBase 向量数据库集成指南
|
||||
|
||||
## 概述
|
||||
|
||||
本文档详细介绍了 OceanBase 向量数据库在 Coze Studio 中的集成适配情况，包括架构设计、实现细节、配置说明和使用指南。
|
||||
|
||||
## 集成背景
|
||||
|
||||
### 为什么选择 OceanBase？
|
||||
|
||||
1. **事务支持**: OceanBase 提供完整的 ACID 事务支持，确保数据一致性
|
||||
2. **部署简单**: 相比 Milvus 等专用向量数据库，OceanBase 部署更简单
|
||||
3. **MySQL 兼容**: 兼容 MySQL 协议，学习成本低
|
||||
4. **向量扩展**: 原生支持向量数据类型和索引
|
||||
5. **运维友好**: 运维成本低，适合中小规模应用
|
||||
|
||||
### 与 Milvus 的对比
|
||||
|
||||
| 特性 | OceanBase | Milvus |
|
||||
| ---------------------- | -------------- | ---------------------- |
|
||||
| **部署复杂度** | 低（单机部署） | 高（需要 etcd、MinIO） |
|
||||
| **事务支持** | 完整 ACID | 有限 |
|
||||
| **向量检索速度** | 中等 | 更快 |
|
||||
| **存储效率** | 中等 | 更高 |
|
||||
| **运维成本** | 低 | 高 |
|
||||
| **学习曲线** | 平缓 | 陡峭 |
|
||||
|
||||
## 架构设计
|
||||
|
||||
### 整体架构
|
||||
|
||||
```
|
||||
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
|
||||
│ Coze Studio │ │ OceanBase │ │ Vector Store │
|
||||
│ Application │───▶│ Client │───▶│ Manager │
|
||||
└─────────────────┘ └─────────────────┘ └─────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────┐
|
||||
│ OceanBase │
|
||||
│ Database │
|
||||
└─────────────────┘
|
||||
```
|
||||
|
||||
### 核心组件
|
||||
|
||||
#### 1. OceanBase Client (`backend/infra/impl/oceanbase/`)
|
||||
|
||||
**主要文件**:
|
||||
|
||||
- `oceanbase.go` - 委托客户端，提供向后兼容接口
|
||||
- `oceanbase_official.go` - 核心实现，基于官方文档
|
||||
- `types.go` - 类型定义
|
||||
|
||||
**核心功能**:
|
||||
|
||||
```go
|
||||
type OceanBaseClient interface {
|
||||
CreateCollection(ctx context.Context, collectionName string) error
|
||||
InsertVectors(ctx context.Context, collectionName string, vectors []VectorResult) error
|
||||
SearchVectors(ctx context.Context, collectionName string, queryVector []float64, topK int) ([]VectorResult, error)
|
||||
DeleteVector(ctx context.Context, collectionName string, vectorID string) error
|
||||
InitDatabase(ctx context.Context) error
|
||||
DropCollection(ctx context.Context, collectionName string) error
|
||||
}
|
||||
```
|
||||
|
||||
#### 2. Search Store Manager (`backend/infra/impl/document/searchstore/oceanbase/`)
|
||||
|
||||
**主要文件**:
|
||||
|
||||
- `oceanbase_manager.go` - 管理器实现
|
||||
- `oceanbase_searchstore.go` - 搜索存储实现
|
||||
- `factory.go` - 工厂模式创建
|
||||
- `consts.go` - 常量定义
|
||||
- `convert.go` - 数据转换
|
||||
- `register.go` - 注册函数
|
||||
|
||||
**核心功能**:
|
||||
|
||||
```go
|
||||
type Manager interface {
|
||||
Create(ctx context.Context, collectionName string) (SearchStore, error)
|
||||
Get(ctx context.Context, collectionName string) (SearchStore, error)
|
||||
Delete(ctx context.Context, collectionName string) error
|
||||
}
|
||||
```
|
||||
|
||||
#### 3. 应用层集成 (`backend/application/base/appinfra/`)
|
||||
|
||||
**文件**: `app_infra.go`
|
||||
|
||||
**集成点**:
|
||||
|
||||
```go
|
||||
case "oceanbase":
|
||||
// 构建 DSN
|
||||
dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local",
|
||||
user, password, host, port, database)
|
||||
|
||||
// 创建客户端
|
||||
client, err := oceanbaseClient.NewOceanBaseClient(dsn)
|
||||
|
||||
// 初始化数据库
|
||||
if err := client.InitDatabase(ctx); err != nil {
|
||||
return nil, fmt.Errorf("init oceanbase database failed, err=%w", err)
|
||||
}
|
||||
```
|
||||
|
||||
## 配置说明
|
||||
|
||||
### 环境变量配置
|
||||
|
||||
#### 必需配置
|
||||
|
||||
```bash
|
||||
# 向量存储类型
|
||||
VECTOR_STORE_TYPE=oceanbase
|
||||
|
||||
# OceanBase 连接配置
|
||||
OCEANBASE_HOST=localhost
|
||||
OCEANBASE_PORT=2881
|
||||
OCEANBASE_USER=root
|
||||
OCEANBASE_PASSWORD=coze123
|
||||
OCEANBASE_DATABASE=test
|
||||
```
|
||||
|
||||
#### 可选配置
|
||||
|
||||
```bash
|
||||
# 性能优化配置
|
||||
OCEANBASE_VECTOR_MEMORY_LIMIT_PERCENTAGE=30
|
||||
OCEANBASE_BATCH_SIZE=100
|
||||
OCEANBASE_MAX_OPEN_CONNS=100
|
||||
OCEANBASE_MAX_IDLE_CONNS=10
|
||||
|
||||
# 缓存配置
|
||||
OCEANBASE_ENABLE_CACHE=true
|
||||
OCEANBASE_CACHE_TTL=300
|
||||
|
||||
# 监控配置
|
||||
OCEANBASE_ENABLE_METRICS=true
|
||||
OCEANBASE_ENABLE_SLOW_QUERY_LOG=true
|
||||
|
||||
# 重试配置
|
||||
OCEANBASE_MAX_RETRIES=3
|
||||
OCEANBASE_RETRY_DELAY=1
|
||||
OCEANBASE_CONN_TIMEOUT=30
|
||||
```
|
||||
|
||||
### Docker 配置
|
||||
|
||||
#### docker-compose-oceanbase.yml
|
||||
|
||||
```yaml
|
||||
oceanbase:
|
||||
image: oceanbase/oceanbase-ce:latest
|
||||
container_name: coze-oceanbase
|
||||
environment:
|
||||
MODE: SLIM
|
||||
OB_DATAFILE_SIZE: 1G
|
||||
OB_SYS_PASSWORD: ${OCEANBASE_PASSWORD:-coze123}
|
||||
OB_TENANT_PASSWORD: ${OCEANBASE_PASSWORD:-coze123}
|
||||
ports:
|
||||
- '2881:2881'
|
||||
volumes:
|
||||
- ./data/oceanbase/ob:/root/ob
|
||||
- ./data/oceanbase/cluster:/root/.obd/cluster
|
||||
deploy:
|
||||
resources:
|
||||
limits:
|
||||
memory: 4G
|
||||
reservations:
|
||||
memory: 2G
|
||||
```
|
||||
|
||||
## 使用指南
|
||||
|
||||
### 1. 快速启动
|
||||
|
||||
```bash
|
||||
# 克隆项目
|
||||
git clone https://github.com/coze-dev/coze-studio.git
|
||||
cd coze-studio
|
||||
|
||||
# 设置 OceanBase 环境文件
|
||||
make oceanbase_env
|
||||
|
||||
# 启动 OceanBase 调试环境
|
||||
make oceanbase_debug
|
||||
```
|
||||
|
||||
### 2. 验证部署
|
||||
|
||||
```bash
|
||||
# 检查容器状态
|
||||
docker ps | grep oceanbase
|
||||
|
||||
# 测试连接
|
||||
mysql -h localhost -P 2881 -u root -p -e "SELECT 1;"
|
||||
|
||||
# 查看数据库
|
||||
mysql -h localhost -P 2881 -u root -p -e "SHOW DATABASES;"
|
||||
```
|
||||
|
||||
### 3. 创建知识库
|
||||
|
||||
在 Coze Studio 界面中：
|
||||
|
||||
1. 进入知识库管理
|
||||
2. 选择 OceanBase 作为向量存储
|
||||
3. 上传文档进行向量化
|
||||
4. 测试向量检索功能
|
||||
|
||||
### 4. 性能监控
|
||||
|
||||
```bash
|
||||
# 查看容器资源使用
|
||||
docker stats coze-oceanbase
|
||||
|
||||
# 查看慢查询日志
|
||||
docker logs coze-oceanbase | grep "slow query"
|
||||
|
||||
# 查看连接数
|
||||
mysql -h localhost -P 2881 -u root -p -e "SHOW PROCESSLIST;"
|
||||
```
|
||||
|
||||
## Helm 部署指南（Kubernetes）
|
||||
|
||||
### 1. 环境准备
|
||||
|
||||
确保已安装以下工具：
|
||||
|
||||
- Kubernetes 集群（推荐使用 k3s 或 kind）
|
||||
- Helm 3.x
|
||||
- kubectl
|
||||
|
||||
### 2. 安装依赖
|
||||
|
||||
#### 安装 cert-manager
|
||||
|
||||
```bash
|
||||
# 添加 cert-manager Helm 仓库
|
||||
helm repo add jetstack https://charts.jetstack.io
|
||||
helm repo update
|
||||
|
||||
# 安装 cert-manager
|
||||
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.16.2/cert-manager.yaml
|
||||
|
||||
# 等待 cert-manager 就绪
|
||||
kubectl wait --for=condition=ready pod -l app.kubernetes.io/name=cert-manager -n cert-manager --timeout=300s
|
||||
```
|
||||
|
||||
#### 安装 ob-operator
|
||||
|
||||
```bash
|
||||
# 添加 ob-operator Helm 仓库
|
||||
helm repo add ob-operator https://oceanbase.github.io/ob-operator/
|
||||
helm repo update
|
||||
|
||||
# 安装 ob-operator
|
||||
helm install ob-operator ob-operator/ob-operator --set reporter=cozeAi --namespace=oceanbase-system --create-namespace
|
||||
|
||||
# 等待 ob-operator 就绪
|
||||
kubectl wait --for=condition=ready pod -l control-plane=controller-manager -n oceanbase-system --timeout=300s
|
||||
```
|
||||
|
||||
### 3. 部署 OceanBase
|
||||
|
||||
#### 使用集成 Helm Chart
|
||||
|
||||
```bash
|
||||
# 部署完整的 Coze Studio 应用（包含 OceanBase）
|
||||
helm install coze-studio helm/charts/opencoze \
|
||||
--set oceanbase.enabled=true \
|
||||
--namespace coze-studio \
|
||||
--create-namespace
|
||||
|
||||
# 或者只部署 OceanBase 组件
|
||||
helm install oceanbase-only helm/charts/opencoze \
|
||||
--set oceanbase.enabled=true \
|
||||
--set mysql.enabled=false \
|
||||
--set redis.enabled=false \
|
||||
--set minio.enabled=false \
|
||||
--set elasticsearch.enabled=false \
|
||||
--set milvus.enabled=false \
|
||||
--set rocketmq.enabled=false \
|
||||
--namespace oceanbase \
|
||||
--create-namespace
|
||||
```
|
||||
|
||||
#### 自定义配置
|
||||
|
||||
创建 `oceanbase-values.yaml` 文件：
|
||||
|
||||
```yaml
|
||||
oceanbase:
|
||||
enabled: true
|
||||
port: 2881
|
||||
targetPort: 2881
|
||||
clusterName: 'cozeAi'
|
||||
clusterId: 1
|
||||
image:
|
||||
repository: oceanbase/oceanbase-ce
|
||||
tag: 'latest'
|
||||
obAgentVersion: '4.2.2-100000042024011120'
|
||||
monitorEnabled: true
|
||||
storageClass: ''
|
||||
observerConfig:
|
||||
resource:
|
||||
cpu: 2
|
||||
memory: 8Gi
|
||||
storages:
|
||||
dataStorage: 10G
|
||||
redoLogStorage: 5G
|
||||
logStorage: 5G
|
||||
monitorResource:
|
||||
cpu: 100m
|
||||
memory: 256Mi
|
||||
generateUserSecrets: true
|
||||
userSecrets:
|
||||
root: 'coze123'
|
||||
monitor: 'coze123'
|
||||
operator: 'coze123'
|
||||
proxyro: 'coze123'
|
||||
topology:
|
||||
- zone: zone1
|
||||
replica: 1
|
||||
parameters:
|
||||
- name: system_memory
|
||||
value: '4G'
|
||||
- name: '__min_full_resource_pool_memory'
|
||||
value: '4294967296'
|
||||
annotations: {}
|
||||
backupVolumeEnabled: false
|
||||
```
|
||||
|
||||
使用自定义配置部署：
|
||||
|
||||
```bash
|
||||
helm install oceanbase-custom helm/charts/opencoze \
|
||||
-f oceanbase-values.yaml \
|
||||
--namespace oceanbase \
|
||||
--create-namespace
|
||||
```
|
||||
|
||||
### 4. 验证部署
|
||||
|
||||
```bash
|
||||
# 检查 OBCluster 状态
|
||||
kubectl get obcluster -n oceanbase
|
||||
|
||||
# 检查 OceanBase pods
|
||||
kubectl get pods -n oceanbase
|
||||
|
||||
# 检查服务
|
||||
kubectl get svc -n oceanbase
|
||||
|
||||
# 查看详细状态
|
||||
kubectl describe obcluster -n oceanbase
|
||||
```
|
||||
|
||||
### 5. 连接测试
|
||||
|
||||
#### 端口转发
|
||||
|
||||
```bash
|
||||
# 转发 OceanBase 端口
|
||||
kubectl port-forward svc/oceanbase-service -n oceanbase 2881:2881
|
||||
```
|
||||
|
||||
#### 使用 obclient 连接
|
||||
|
||||
```bash
|
||||
# 在集群内连接
|
||||
kubectl exec -it deployment/oceanbase-obcluster-zone1 -n oceanbase -- obclient -h127.0.0.1 -P2881 -uroot@test -pcoze123 -Dtest
|
||||
|
||||
# 从外部连接（需要端口转发）
|
||||
obclient -h127.0.0.1 -P2881 -uroot@test -pcoze123 -Dtest
|
||||
```
|
||||
|
||||
#### 使用 MySQL 客户端连接
|
||||
|
||||
```bash
|
||||
# 使用 MySQL 客户端
|
||||
mysql -h127.0.0.1 -P2881 -uroot@test -pcoze123 -Dtest
|
||||
```
|
||||
|
||||
### 6. 监控和管理
|
||||
|
||||
#### 查看日志
|
||||
|
||||
```bash
|
||||
# 查看 OceanBase 日志
|
||||
kubectl logs -f deployment/oceanbase-obcluster-zone1 -n oceanbase
|
||||
|
||||
# 查看 ob-operator 日志
|
||||
kubectl logs -f deployment/oceanbase-controller-manager -n oceanbase-system
|
||||
```
|
||||
|
||||
#### 扩缩容
|
||||
|
||||
```bash
|
||||
# 扩展副本数
|
||||
kubectl patch obcluster oceanbase-obcluster -n oceanbase --type='merge' -p='{"spec":{"topology":[{"zone":"zone1","replica":2}]}}'
|
||||
|
||||
# 调整资源配置
|
||||
kubectl patch obcluster oceanbase-obcluster -n oceanbase --type='merge' -p='{"spec":{"observer":{"resource":{"cpu":4,"memory":"16Gi"}}}}'
|
||||
```
|
||||
|
||||
#### 备份和恢复
|
||||
|
||||
```bash
|
||||
# 创建备份
|
||||
kubectl apply -f - <<EOF
|
||||
apiVersion: oceanbase.oceanbase.com/v1alpha1
|
||||
kind: OBTenantBackupPolicy
|
||||
metadata:
|
||||
name: backup-policy
|
||||
namespace: oceanbase
|
||||
spec:
|
||||
obClusterName: oceanbase-obcluster
|
||||
tenantName: test
|
||||
backupType: FULL
|
||||
schedule: "0 2 * * *"
|
||||
destination:
|
||||
path: "file:///backup"
|
||||
EOF
|
||||
```
|
||||
|
||||
### 7. 故障排除
|
||||
|
||||
#### 常见问题
|
||||
|
||||
1. **OBCluster 创建失败**
|
||||
|
||||
```bash
|
||||
# 检查 ob-operator 状态
|
||||
kubectl get pods -n oceanbase-system
|
||||
|
||||
# 查看详细错误
|
||||
kubectl describe obcluster -n oceanbase
|
||||
```
|
||||
2. **镜像拉取失败**
|
||||
|
||||
```bash
|
||||
# 检查节点镜像拉取能力
|
||||
kubectl describe node
|
||||
|
||||
# 手动拉取镜像
|
||||
docker pull oceanbase/oceanbase-cloud-native:4.3.5.3-103000092025080818
|
||||
```
|
||||
3. **存储问题**
|
||||
|
||||
```bash
|
||||
# 检查 PVC 状态
|
||||
kubectl get pvc -n oceanbase
|
||||
|
||||
# 检查存储类
|
||||
kubectl get storageclass
|
||||
```
|
||||
|
||||
#### 日志分析
|
||||
|
||||
```bash
|
||||
# 查看所有相关日志
|
||||
kubectl logs -f deployment/oceanbase-controller-manager -n oceanbase-system
|
||||
kubectl logs -f deployment/oceanbase-obcluster-zone1 -n oceanbase
|
||||
kubectl logs -f deployment/cert-manager -n cert-manager
|
||||
```
|
||||
|
||||
### 8. 卸载
|
||||
|
||||
```bash
|
||||
# 卸载 OceanBase
|
||||
helm uninstall oceanbase-custom -n oceanbase
|
||||
|
||||
# 删除 namespace
|
||||
kubectl delete namespace oceanbase
|
||||
|
||||
# 卸载 ob-operator
|
||||
helm uninstall ob-operator -n oceanbase-system
|
||||
|
||||
# 卸载 cert-manager
|
||||
kubectl delete -f https://github.com/cert-manager/cert-manager/releases/download/v1.16.2/cert-manager.yaml
|
||||
```
|
||||
|
||||
## 适配特点
|
||||
|
||||
### 1. 设计原则
|
||||
|
||||
#### 架构兼容性设计
|
||||
|
||||
- 严格遵循 Coze Studio 核心架构设计原则，确保 OceanBase 适配层与现有系统无缝集成
|
||||
- 采用委托模式（Delegation Pattern）实现向后兼容，保证现有接口的稳定性和一致性
|
||||
- 保持与现有向量存储接口的完全兼容，确保系统平滑迁移和升级
|
||||
|
||||
#### 性能优先
|
||||
|
||||
- 使用 HNSW 索引实现高效的近似最近邻搜索
|
||||
- 批量操作减少数据库交互次数
|
||||
- 连接池管理优化资源使用
|
||||
|
||||
#### 易于部署
|
||||
|
||||
- 单机部署，无需复杂的集群配置
|
||||
- Docker 一键部署
|
||||
- 环境变量配置，灵活易用
|
||||
|
||||
### 2. 技术亮点
|
||||
|
||||
#### 委托模式设计
|
||||
|
||||
```go
|
||||
type OceanBaseClient struct {
|
||||
official *OceanBaseOfficialClient
|
||||
}
|
||||
|
||||
func (c *OceanBaseClient) CreateCollection(ctx context.Context, collectionName string) error {
|
||||
return c.official.CreateCollection(ctx, collectionName)
|
||||
}
|
||||
```
|
||||
|
||||
#### 智能配置管理
|
||||
|
||||
```go
|
||||
func DefaultConfig() *Config {
|
||||
return &Config{
|
||||
Host: getEnv("OCEANBASE_HOST", "localhost"),
|
||||
Port: getEnvAsInt("OCEANBASE_PORT", 2881),
|
||||
User: getEnv("OCEANBASE_USER", "root"),
|
||||
Password: getEnv("OCEANBASE_PASSWORD", ""),
|
||||
Database: getEnv("OCEANBASE_DATABASE", "test"),
|
||||
// ... 其他配置
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### 错误处理优化
|
||||
|
||||
```go
|
||||
func (c *OceanBaseOfficialClient) setVectorParameters() error {
|
||||
params := map[string]string{
|
||||
"ob_vector_memory_limit_percentage": "30",
|
||||
"ob_query_timeout": "86400000000",
|
||||
"max_allowed_packet": "1073741824",
|
||||
}
|
||||
|
||||
for param, value := range params {
|
||||
if err := c.db.Exec(fmt.Sprintf("SET GLOBAL %s = %s", param, value)).Error; err != nil {
|
||||
log.Printf("Warning: Failed to set %s: %v", param, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
## 故障排查
|
||||
|
||||
### 1. 常见问题
|
||||
|
||||
#### 连接问题
|
||||
|
||||
```bash
|
||||
# 检查容器状态
|
||||
docker ps | grep oceanbase
|
||||
|
||||
# 检查端口映射
|
||||
docker port coze-oceanbase
|
||||
|
||||
# 测试连接
|
||||
mysql -h localhost -P 2881 -u root -p -e "SELECT 1;"
|
||||
```
|
||||
|
||||
#### 向量索引问题
|
||||
|
||||
```sql
|
||||
-- 检查索引状态
|
||||
SHOW INDEX FROM test_vectors;
|
||||
|
||||
-- 重建索引
|
||||
DROP INDEX idx_test_embedding ON test_vectors;
|
||||
CREATE VECTOR INDEX idx_test_embedding ON test_vectors(embedding)
|
||||
WITH (distance=cosine, type=hnsw, lib=vsag, m=16, ef_construction=200, ef_search=64);
|
||||
```
|
||||
|
||||
#### 性能问题
|
||||
|
||||
```sql
|
||||
-- 调整内存限制
|
||||
SET GLOBAL ob_vector_memory_limit_percentage = 50;
|
||||
|
||||
-- 查看慢查询
|
||||
SHOW VARIABLES LIKE 'slow_query_log';
|
||||
```
|
||||
|
||||
### 2. 日志分析
|
||||
|
||||
```bash
|
||||
# 查看 OceanBase 日志
|
||||
docker logs coze-oceanbase
|
||||
|
||||
# 查看应用日志
|
||||
tail -f logs/coze-studio.log | grep -i "oceanbase\|vector"
|
||||
```
|
||||
|
||||
## 总结
|
||||
|
||||
OceanBase 向量数据库在 Coze Studio 中的集成实现了以下目标：
|
||||
|
||||
1. **功能完整**: 支持完整的向量存储和检索功能
|
||||
2. **性能良好**: 通过 HNSW 索引实现高效的向量搜索
|
||||
3. **部署简单**: 单机部署，无需复杂配置
|
||||
4. **运维友好**: 低运维成本，易于监控和管理
|
||||
5. **扩展性强**: 支持水平扩展和垂直扩展
|
||||
|
||||
通过这次集成，Coze Studio 为用户提供了一个简单、高效、可靠的向量数据库解决方案，特别适合需要事务支持、部署简单、运维成本低的场景。
|
||||
|
||||
## 相关链接
|
||||
|
||||
- [OceanBase 官方文档](https://www.oceanbase.com/docs)
|
||||
- [Coze Studio 项目地址](https://github.com/coze-dev/coze-studio)
|
||||
@@ -0,0 +1,470 @@
|
||||
# Pulsar EventBus Integration Guide
|
||||
|
||||
## Overview
|
||||
|
||||
This document provides a comprehensive guide for integrating Apache Pulsar as an EventBus in Coze Studio, including architecture design, implementation details, configuration instructions, and usage guidelines.
|
||||
|
||||
## Integration Background
|
||||
|
||||
### Why Choose Pulsar?
|
||||
|
||||
In Coze Studio's architecture, EventBus plays a critical role in asynchronous message delivery, including workflow execution, Agent communication, data processing pipelines, and other core functions. As user scale grows and business complexity increases, we need a more powerful and flexible message queue solution.
|
||||
|
||||
Pulsar, as a next-generation distributed messaging system, brings the following core advantages to Coze Studio:
|
||||
|
||||
1. **High Performance**: Pulsar provides low-latency, high-throughput messaging that can support Coze Studio's large-scale concurrent Agent execution and workflow processing
|
||||
2. **Multi-tenancy**: Native support for multi-tenant architecture, perfectly matching Coze Studio's multi-user, multi-workspace business model
|
||||
3. **Persistence**: Supports message persistence storage, ensuring the reliability of Agent execution states and workflow data, preventing task loss due to system restarts
|
||||
4. **Horizontal Scaling**: Supports separation of compute and storage, easy to scale horizontally, enabling smooth scaling as Coze Studio's user base grows
|
||||
5. **Message Ordering**: Pulsar provides strong consistency and message ordering guarantees, ensuring that Agent workflow steps execute in the correct sequence, preventing state confusion and data inconsistency
|
||||
6. **Rich Features**: Supports message deduplication, delayed messages, dead letter queues, and other advanced features, providing stronger reliability guarantees for complex AI workflows
|
||||
|
||||
### Comparison with Other MQ Systems
|
||||
|
||||
| Feature | Pulsar | NSQ | Kafka | RocketMQ |
|
||||
| ---------------------- | -------------- | -------------- | -------------- | -------------- |
|
||||
| **Deployment Complexity** | Medium | Low | Medium | Medium |
|
||||
| **Performance** | High | Medium | High | High |
|
||||
| **Multi-tenancy** | Native Support | Not Supported | Limited | Limited |
|
||||
| **Message Persistence** | Strong | Limited | Strong | Strong |
|
||||
| **Message Ordering** | Strong | Weak | Strong | Strong |
|
||||
| **Horizontal Scaling** | Excellent | Medium | Good | Good |
|
||||
| **Scaling Speed** | Fast | Medium | Slow | Medium |
|
||||
| **Operational Complexity** | Medium | Low | High | Medium |
|
||||
| **Ecosystem** | Rich | Simple | Very Rich | Rich |
|
||||
|
||||
#### Detailed Comparison of Horizontal Scaling Capabilities
|
||||
|
||||
**Pulsar's Scaling Advantages**:
|
||||
- **Compute-Storage Separation**: Broker (compute) and BookKeeper (storage) scale independently, allowing precise resource adjustment based on business needs
|
||||
- **Stateless Brokers**: Broker nodes are stateless and can start/stop quickly, enabling second-level scaling
|
||||
- **Automatic Load Balancing**: Automatic redistribution of Topics and Partitions when new Brokers are added
|
||||
- **Hot Scaling**: Supports dynamic addition/removal of nodes without service interruption
|
||||
|
||||
**Comparison with Other MQ Systems**:
|
||||
- **Kafka**: Requires manual Partition rebalancing, complex and time-consuming scaling process that may affect business operations
|
||||
- **RocketMQ**: While supporting dynamic scaling, the coordination mechanism between NameServer and Broker is relatively complex
|
||||
- **NSQ**: Single-machine architecture limits scaling capabilities, can only improve throughput by increasing Topic count
|
||||
|
||||
This excellent scaling capability makes Pulsar particularly suitable for scenarios like Coze Studio with rapid user growth and fluctuating business loads.
|
||||
|
||||
## Architecture Design
|
||||
|
||||
### Overall Architecture
|
||||
|
||||
```
|
||||
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
|
||||
│ Coze Studio │ │ Pulsar │ │ EventBus │
|
||||
│ Application │───▶│ Client │───▶│ Manager │
|
||||
└─────────────────┘ └─────────────────┘ └─────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────┐
|
||||
│ Apache Pulsar │
|
||||
│ Cluster │
|
||||
└─────────────────┘
|
||||
```
|
||||
|
||||
### Core Components
|
||||
|
||||
#### 1. Pulsar Producer
|
||||
|
||||
**File Location**: `backend/infra/impl/eventbus/pulsar/producer.go`
|
||||
|
||||
**Core Functions**:
|
||||
|
||||
```go
|
||||
type Producer interface {
|
||||
Send(ctx context.Context, body []byte, opts ...SendOpt) error
|
||||
BatchSend(ctx context.Context, bodyArr [][]byte, opts ...SendOpt) error
|
||||
}
|
||||
|
||||
type producerImpl struct {
|
||||
topic string
|
||||
client pulsar.Client
|
||||
producer pulsar.Producer
|
||||
}
|
||||
```
|
||||
|
||||
**Features**:
|
||||
- Supports synchronous and asynchronous sending
|
||||
- Batch sending for performance optimization
|
||||
- JWT authentication support
|
||||
- Graceful shutdown handling
|
||||
|
||||
#### 2. Pulsar Consumer
|
||||
|
||||
**File Location**: `backend/infra/impl/eventbus/pulsar/consumer.go`
|
||||
|
||||
**Core Functions**:
|
||||
|
||||
```go
|
||||
func RegisterConsumer(serviceURL, topic, group string,
|
||||
consumerHandler eventbus.ConsumerHandler,
|
||||
opts ...eventbus.ConsumerOpt) error
|
||||
```
|
||||
|
||||
**Features**:
|
||||
- Exclusive mode consumption for message ordering
|
||||
- Automatic retry and error handling
|
||||
- Context cancellation support
|
||||
- Message acknowledgment and negative acknowledgment mechanisms
|
||||
|
||||
#### 3. EventBus Factory
|
||||
|
||||
**File Location**: `backend/infra/impl/eventbus/eventbus.go`
|
||||
|
||||
**Integration Point**:
|
||||
|
||||
```go
|
||||
case consts.MQTypePulsar:
|
||||
return pulsar.NewProducer(nameServer, topic, group)
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
#### Required Configuration
|
||||
|
||||
```bash
|
||||
# Message queue type
|
||||
COZE_MQ_TYPE=pulsar
|
||||
|
||||
# Pulsar service address
|
||||
MQ_NAME_SERVER=pulsar://localhost:6650
|
||||
```
|
||||
|
||||
#### Optional Configuration
|
||||
|
||||
```bash
|
||||
# JWT authentication token (if authentication is enabled)
|
||||
PULSAR_JWT_TOKEN=eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.example_token
|
||||
```
|
||||
|
||||
### Docker Configuration
|
||||
|
||||
#### Standalone Pulsar Deployment
|
||||
|
||||
```yaml
|
||||
services:
|
||||
pulsar:
|
||||
image: apachepulsar/pulsar:3.0.12
|
||||
container_name: coze-pulsar
|
||||
restart: always
|
||||
command: >
|
||||
sh -c "bin/pulsar standalone"
|
||||
ports:
|
||||
- "6650:6650" # Pulsar service port
|
||||
- "8080:8080" # Pulsar admin port
|
||||
volumes:
|
||||
- ./data/pulsar:/pulsar/data
|
||||
networks:
|
||||
- coze-network
|
||||
healthcheck:
|
||||
test: ["CMD", "curl", "-f", "http://localhost:8080/admin/v2/clusters"]
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 5
|
||||
start_period: 30s
|
||||
```
|
||||
|
||||
#### Production Cluster Deployment
|
||||
|
||||
For production environments, it's recommended to use Pulsar cluster deployment to achieve high availability and better performance. Cluster deployment involves configuring multiple components including ZooKeeper, BookKeeper, and Broker, which can be quite complex.
|
||||
|
||||
**Production Environment Recommendations**:
|
||||
- Use Pulsar cluster mode deployment for high availability
|
||||
- Enable JWT authentication for security
|
||||
- Configure appropriate resource limits and monitoring
|
||||
|
||||
For detailed cluster deployment configuration, please refer to the [Apache Pulsar Official Documentation](https://pulsar.apache.org/docs/4.1.x/deploy-kubernetes/).
|
||||
|
||||
## Usage Guide
|
||||
|
||||
### 1. Prepare Project
|
||||
|
||||
```bash
|
||||
# Clone the project
|
||||
git clone https://github.com/coze-dev/coze-studio.git
|
||||
cd coze-studio
|
||||
```
|
||||
|
||||
### 2. Modify Docker Compose Configuration
|
||||
|
||||
Add the Pulsar service to your `docker/docker-compose.yml` file:
|
||||
|
||||
```yaml
|
||||
services:
|
||||
# Add Pulsar service
|
||||
pulsar:
|
||||
image: apachepulsar/pulsar:3.0.12
|
||||
container_name: coze-pulsar
|
||||
restart: always
|
||||
command: >
|
||||
sh -c "bin/pulsar standalone"
|
||||
ports:
|
||||
- "6650:6650" # Pulsar service port
|
||||
- "8080:8080" # Pulsar admin port
|
||||
volumes:
|
||||
- ./data/pulsar:/pulsar/data
|
||||
networks:
|
||||
- coze-network
|
||||
healthcheck:
|
||||
test: ["CMD", "curl", "-f", "http://localhost:8080/admin/v2/clusters"]
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 5
|
||||
start_period: 30s
|
||||
|
||||
# Other existing services...
|
||||
```
|
||||
|
||||
### 3. Configure Environment Variables
|
||||
|
||||
Modify the `.env` file to configure Coze Studio to use Pulsar:
|
||||
|
||||
```bash
|
||||
# Enter docker directory
|
||||
cd docker
|
||||
|
||||
# Copy environment configuration file
|
||||
cp .env.example .env
|
||||
|
||||
# Edit .env file and add the following configuration:
|
||||
# Message queue type
|
||||
COZE_MQ_TYPE=pulsar
|
||||
|
||||
# Pulsar service address
|
||||
MQ_NAME_SERVER=pulsar://pulsar:6650
|
||||
|
||||
# JWT authentication token (optional, if authentication is enabled)
|
||||
# PULSAR_JWT_TOKEN=your_jwt_token_here
|
||||
```
|
||||
|
||||
### 4. Start Services
|
||||
|
||||
```bash
|
||||
# Start complete Coze Studio services including Pulsar
|
||||
docker-compose up -d
|
||||
|
||||
# Check service startup status
|
||||
docker-compose ps
|
||||
```
|
||||
|
||||
### 5. Verify Deployment
|
||||
|
||||
```bash
|
||||
# Check Pulsar container status
|
||||
docker ps | grep pulsar
|
||||
|
||||
# Check Pulsar health status
|
||||
curl -f http://localhost:8080/admin/v2/clusters
|
||||
|
||||
# View Pulsar logs
|
||||
docker logs coze-pulsar
|
||||
|
||||
# Test Pulsar connection
|
||||
docker exec -it coze-pulsar bin/pulsar-admin clusters list
|
||||
```
|
||||
|
||||
### 6. Access Services
|
||||
|
||||
- **Coze Studio**: `http://localhost:3000` (based on actual configuration)
|
||||
- **Pulsar Admin**: `http://localhost:8080`
|
||||
|
||||
Now Coze Studio has successfully integrated Pulsar as the message queue, and all EventBus functionality will be handled through Pulsar.
|
||||
|
||||
## Appendix
|
||||
|
||||
### A. Production Cluster Deployment
|
||||
|
||||
For production environments, it's recommended to use Pulsar cluster deployment to achieve high availability and better performance. Cluster deployment involves configuring multiple components including ZooKeeper, BookKeeper, and Broker, which can be quite complex.
|
||||
|
||||
**Production Environment Recommendations**:
|
||||
- Use Pulsar cluster mode deployment for high availability
|
||||
- Enable JWT authentication for security
|
||||
- Configure appropriate resource limits and monitoring
|
||||
|
||||
For detailed cluster deployment configuration, please refer to the [Apache Pulsar Official Documentation](https://pulsar.apache.org/docs/4.1.x/deploy-kubernetes/).
|
||||
|
||||
### B. Visual Management Tools
|
||||
|
||||
For users who need a graphical interface to manage Pulsar clusters, consider using ASP Community Edition. ASP Community Edition is a modern management platform designed specifically for Apache Pulsar, providing an intuitive web interface to manage clusters, tenants, namespaces, topics, and other resources. The platform supports real-time monitoring, performance metrics display, configuration management, and other features that greatly simplify the daily operations of Pulsar clusters.
|
||||
|
||||
For more information, please refer to: [ASP Community Edition Documentation](https://ascentstream.com/docs/asp/asp-community/overview)
|
||||
|
||||
### C. Integration Features
|
||||
|
||||
#### 1. Design Principles
|
||||
|
||||
**Architecture Compatibility Design**:
|
||||
- Strictly follows Coze Studio EventBus interface specifications for seamless integration
|
||||
- Uses factory pattern for unified management of multiple MQ systems
|
||||
- Maintains interface consistency with NSQ, Kafka, and RocketMQ implementations
|
||||
|
||||
**Performance First**:
|
||||
- Asynchronous batch sending reduces network overhead
|
||||
- Connection pooling reduces connection costs
|
||||
- Message acknowledgment mechanism ensures reliability
|
||||
|
||||
**Easy Deployment**:
|
||||
- Standalone mode for quick startup
|
||||
- Docker containerized deployment
|
||||
- Environment variable configuration for flexibility
|
||||
|
||||
#### 2. Technical Highlights
|
||||
|
||||
**JWT Authentication Support**:
|
||||
```go
|
||||
// Automatically detect and configure JWT authentication
|
||||
if jwtToken := os.Getenv(consts.PulsarJWTToken); jwtToken != "" {
|
||||
clientOptions.Authentication = pulsar.NewAuthenticationToken(jwtToken)
|
||||
logs.Debugf("Using JWT authentication, token length: %d", len(jwtToken))
|
||||
}
|
||||
```
|
||||
|
||||
**Batch Sending Optimization**:
|
||||
```go
|
||||
// Asynchronous batch sending for improved performance
|
||||
for _, body := range bodyArr {
|
||||
msg := &pulsar.ProducerMessage{Payload: body}
|
||||
if option.ShardingKey != nil {
|
||||
msg.Key = *option.ShardingKey
|
||||
}
|
||||
p.producer.SendAsync(ctx, msg, callback)
|
||||
}
|
||||
```
|
||||
|
||||
**Graceful Shutdown Handling**:
|
||||
```go
|
||||
// Listen for system signals and gracefully close resources
|
||||
safego.Go(context.Background(), func() {
|
||||
signal.WaitExit()
|
||||
logs.Infof("shutting down pulsar consumer for topic: %s, group: %s", topic, group)
|
||||
cancel()
|
||||
consumer.Close()
|
||||
client.Close()
|
||||
})
|
||||
```
|
||||
|
||||
### C. Troubleshooting
|
||||
|
||||
#### 1. Common Issues
|
||||
|
||||
**Connection Issues**:
|
||||
```bash
|
||||
# Check Pulsar service status
|
||||
docker exec -it coze-pulsar bin/pulsar-admin brokers healthcheck
|
||||
|
||||
# Check network connectivity
|
||||
telnet localhost 6650
|
||||
|
||||
# View connection configuration
|
||||
docker exec -it coze-pulsar cat conf/standalone.conf | grep -E "(advertisedAddress|bindAddress)"
|
||||
```
|
||||
|
||||
**Authentication Issues**:
|
||||
```bash
|
||||
# Check JWT Token configuration
|
||||
echo $PULSAR_JWT_TOKEN
|
||||
|
||||
# Verify token validity
|
||||
docker exec -it coze-pulsar bin/pulsar-admin --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
|
||||
--auth-params token:$PULSAR_JWT_TOKEN \
|
||||
clusters list
|
||||
```
|
||||
|
||||
**Performance Issues**:
|
||||
```bash
|
||||
# View topic backlog
|
||||
docker exec -it coze-pulsar bin/pulsar-admin topics stats persistent://public/default/your-topic
|
||||
|
||||
# Adjust batch sending parameters
|
||||
# Batch size and delay can be configured through SendOpt in code
|
||||
```
|
||||
|
||||
#### 2. Log Analysis
|
||||
|
||||
```bash
|
||||
# View Pulsar service logs
|
||||
docker logs coze-pulsar
|
||||
|
||||
# View Pulsar-related information in application logs
|
||||
tail -f logs/coze-studio.log | grep -i "pulsar\|eventbus"
|
||||
|
||||
# Enable verbose logging
|
||||
# Set rootLogLevel=DEBUG in Pulsar configuration
|
||||
```
|
||||
|
||||
#### 3. Monitoring Metrics
|
||||
|
||||
```bash
|
||||
# Get broker metrics
|
||||
curl http://localhost:8080/metrics/
|
||||
|
||||
# Get specific topic metrics
|
||||
curl http://localhost:8080/admin/v2/persistent/public/default/your-topic/stats
|
||||
|
||||
# Monitor consumption lag
|
||||
docker exec -it coze-pulsar bin/pulsar-admin topics subscriptions persistent://public/default/your-topic
|
||||
```
|
||||
|
||||
### D. Best Practices
|
||||
|
||||
#### 1. Production Environment Configuration
|
||||
|
||||
```bash
|
||||
# Recommended production environment configuration
|
||||
COZE_MQ_TYPE=pulsar
|
||||
MQ_NAME_SERVER=pulsar://pulsar-broker-1:6650,pulsar://pulsar-broker-2:6650,pulsar://pulsar-broker-3:6650
|
||||
PULSAR_JWT_TOKEN=your-production-jwt-token
|
||||
|
||||
# Pulsar cluster configuration
|
||||
# Recommend at least 3 Broker nodes
|
||||
# Recommend at least 3 BookKeeper nodes
|
||||
# Recommend at least 3 ZooKeeper nodes
|
||||
```
|
||||
|
||||
#### 2. Performance Tuning
|
||||
|
||||
```bash
|
||||
# Producer configuration optimization
|
||||
# Batch size: 1000 messages or 1MB
|
||||
# Send timeout: 30 seconds
|
||||
# Compression algorithm: LZ4
|
||||
|
||||
# Consumer configuration optimization
|
||||
# Receive queue size: 1000
|
||||
# Acknowledgment timeout: 30 seconds
|
||||
# Consumer type: Exclusive (ensures ordering)
|
||||
```
|
||||
|
||||
#### 3. Security Configuration
|
||||
|
||||
```bash
|
||||
# Enable JWT authentication
|
||||
PULSAR_JWT_TOKEN=your-jwt-token
|
||||
|
||||
# Configure Access Control Lists (ACL)
|
||||
# Configure topic-level permissions through Pulsar Admin tools
|
||||
```
|
||||
|
||||
## Summary
|
||||
|
||||
The Apache Pulsar EventBus integration in Coze Studio achieves the following goals:
|
||||
|
||||
1. **High Performance**: Supports high-throughput, low-latency messaging
|
||||
2. **High Reliability**: Message persistence storage with acknowledgment mechanisms
|
||||
3. **Easy Scaling**: Supports horizontal scaling to accommodate business growth
|
||||
4. **Easy Operations**: Rich management tools and monitoring metrics
|
||||
5. **Enterprise-grade**: Multi-tenancy support for enterprise applications
|
||||
|
||||
Through this integration, Coze Studio provides users with a high-performance, highly reliable, and easily scalable message queue solution, particularly suitable for scenarios requiring high throughput, low latency, and enterprise-grade features.
|
||||
|
||||
## Related Links
|
||||
|
||||
- [Apache Pulsar Official Documentation](https://pulsar.apache.org/docs/)
|
||||
- [Pulsar Go Client Documentation](https://pulsar.apache.org/docs/client-libraries-go/)
|
||||
- [ASP Community Edition Documentation](https://ascentstream.com/docs/asp/asp-community/overview)
|
||||
- [Coze Studio Project Repository](https://github.com/coze-dev/coze-studio)
|
||||
@@ -0,0 +1,411 @@
|
||||
# Pulsar EventBus 集成指南
|
||||
|
||||
## 概述
|
||||
|
||||
本文档详细介绍了 Apache Pulsar 作为 EventBus 在 Coze Studio 中的集成适配情况，包括架构设计、实现细节、配置说明和使用指南。
|
||||
|
||||
## 集成背景
|
||||
|
||||
### 为什么选择 Pulsar？
|
||||
|
||||
在 Coze Studio 的架构中，EventBus 承担着关键的异步消息传递任务，包括工作流执行、Agent 通信、数据处理管道等核心功能。随着用户规模的增长和业务复杂度的提升，我们需要一个更加强大和灵活的消息队列解决方案。
|
||||
|
||||
Pulsar 作为新一代的分布式消息系统，为 Coze Studio 带来了以下核心优势：
|
||||
|
||||
1. **高性能**: Pulsar 提供低延迟、高吞吐量的消息传递，能够支撑 Coze Studio 大规模并发的 Agent 执行和工作流处理
|
||||
2. **多租户**: 原生支持多租户架构，完美契合 Coze Studio 多用户、多工作空间的业务模式
|
||||
3. **持久化**: 支持消息持久化存储，确保 Agent 执行状态和工作流数据的可靠性，避免因系统重启导致的任务丢失
|
||||
4. **水平扩展**: 支持计算和存储分离，易于水平扩展，能够随着 Coze Studio 用户增长而平滑扩容
|
||||
5. **顺序性保障**: Pulsar 提供强一致性的消息顺序保证，确保 Agent 工作流中的步骤按正确顺序执行，避免状态混乱和数据不一致
|
||||
6. **丰富特性**: 支持消息去重、延迟消息、死信队列等高级特性，为复杂的 AI 工作流提供更强的可靠性保障
|
||||
|
||||
### 与其他 MQ 的对比
|
||||
|
||||
| 特性 | Pulsar | NSQ | Kafka | RocketMQ |
|
||||
| ---------------------- | -------------- | -------------- | -------------- | -------------- |
|
||||
| **部署复杂度** | 中等 | 低 | 中等 | 中等 |
|
||||
| **性能** | 高 | 中等 | 高 | 高 |
|
||||
| **多租户支持** | 原生支持 | 不支持 | 有限支持 | 有限支持 |
|
||||
| **消息持久化** | 强 | 有限 | 强 | 强 |
|
||||
| **顺序性保障** | 强 | 弱 | 强 | 强 |
|
||||
| **水平扩展性** | 优秀 | 中等 | 良好 | 良好 |
|
||||
| **扩缩容速度** | 快速 | 中等 | 慢 | 中等 |
|
||||
| **运维复杂度** | 中等 | 低 | 高 | 中等 |
|
||||
| **生态系统** | 丰富 | 简单 | 非常丰富 | 丰富 |
|
||||
|
||||
#### 水平扩展能力详细对比
|
||||
|
||||
**Pulsar 的扩展优势**：
|
||||
- **计算存储分离**：Broker（计算）和 BookKeeper（存储）独立扩展，可以根据业务需求精确调整资源
|
||||
- **无状态 Broker**：Broker 节点无状态，可以快速启动和停止，实现秒级扩缩容
|
||||
- **自动负载均衡**：新增 Broker 后自动进行 Topic 和 Partition 的负载重分配
|
||||
- **热扩容**：支持在不停服的情况下动态增减节点，对业务无影响
|
||||
|
||||
**与其他 MQ 的对比**：
|
||||
- **Kafka**：需要手动进行 Partition 重分配，扩容过程复杂且耗时，可能影响业务
|
||||
- **RocketMQ**：虽然支持动态扩容，但 NameServer 和 Broker 的协调机制相对复杂
|
||||
- **NSQ**：单机架构限制了扩展能力，只能通过增加 Topic 数量来提升吞吐量
|
||||
|
||||
这种优秀的扩展能力使得 Pulsar 特别适合 Coze Studio 这种用户增长快速、业务负载波动大的场景。
|
||||
|
||||
## 架构设计
|
||||
|
||||
### 整体架构
|
||||
|
||||
```
|
||||
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
|
||||
│ Coze Studio │ │ Pulsar │ │ EventBus │
|
||||
│ Application │───▶│ Client │───▶│ Manager │
|
||||
└─────────────────┘ └─────────────────┘ └─────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────┐
|
||||
│ Apache Pulsar │
|
||||
│ Cluster │
|
||||
└─────────────────┘
|
||||
```
|
||||
|
||||
### 核心组件
|
||||
|
||||
#### 1. Pulsar Producer
|
||||
|
||||
**文件位置**: `backend/infra/impl/eventbus/pulsar/producer.go`
|
||||
|
||||
**核心功能**:
|
||||
|
||||
```go
|
||||
type Producer interface {
|
||||
Send(ctx context.Context, body []byte, opts ...SendOpt) error
|
||||
BatchSend(ctx context.Context, bodyArr [][]byte, opts ...SendOpt) error
|
||||
}
|
||||
|
||||
type producerImpl struct {
|
||||
topic string
|
||||
client pulsar.Client
|
||||
producer pulsar.Producer
|
||||
}
|
||||
```
|
||||
|
||||
**特性**:
|
||||
- 支持同步和异步发送
|
||||
- 批量发送优化性能
|
||||
- JWT 认证支持
|
||||
- 优雅关闭处理
|
||||
|
||||
#### 2. Pulsar Consumer
|
||||
|
||||
**文件位置**: `backend/infra/impl/eventbus/pulsar/consumer.go`
|
||||
|
||||
**核心功能**:
|
||||
|
||||
```go
|
||||
func RegisterConsumer(serviceURL, topic, group string,
|
||||
consumerHandler eventbus.ConsumerHandler,
|
||||
opts ...eventbus.ConsumerOpt) error
|
||||
```
|
||||
|
||||
**特性**:
|
||||
- 独占模式消费，保证消息顺序
|
||||
- 自动重试和错误处理
|
||||
- 上下文取消支持
|
||||
- 消息确认和否认机制
|
||||
|
||||
#### 3. EventBus 工厂
|
||||
|
||||
**文件位置**: `backend/infra/impl/eventbus/eventbus.go`
|
||||
|
||||
**集成点**:
|
||||
|
||||
```go
|
||||
case consts.MQTypePulsar:
|
||||
return pulsar.NewProducer(nameServer, topic, group)
|
||||
```
|
||||
|
||||
## 使用指南
|
||||
|
||||
### 1. 准备项目
|
||||
|
||||
```bash
|
||||
# 克隆项目
|
||||
git clone https://github.com/coze-dev/coze-studio.git
|
||||
cd coze-studio
|
||||
```
|
||||
|
||||
### 2. 修改 Docker Compose 配置
|
||||
|
||||
在 `docker/docker-compose.yml` 文件中添加 Pulsar 服务：
|
||||
|
||||
```yaml
|
||||
services:
|
||||
# 添加 Pulsar 服务
|
||||
pulsar:
|
||||
image: apachepulsar/pulsar:3.0.12
|
||||
container_name: coze-pulsar
|
||||
restart: always
|
||||
command: >
|
||||
sh -c "bin/pulsar standalone"
|
||||
ports:
|
||||
- "6650:6650" # Pulsar 服务端口
|
||||
- "8080:8080" # Pulsar 管理端口
|
||||
volumes:
|
||||
- ./data/pulsar:/pulsar/data
|
||||
networks:
|
||||
- coze-network
|
||||
healthcheck:
|
||||
test: ["CMD", "curl", "-f", "http://localhost:8080/admin/v2/clusters"]
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 5
|
||||
start_period: 30s
|
||||
|
||||
# 其他现有服务...
|
||||
```
|
||||
|
||||
### 3. 配置环境变量
|
||||
|
||||
修改 `.env` 文件，配置 Coze Studio 使用 Pulsar：
|
||||
|
||||
```bash
|
||||
# 进入 docker 目录
|
||||
cd docker
|
||||
|
||||
# 复制环境配置文件
|
||||
cp .env.example .env
|
||||
|
||||
# 编辑 .env 文件，添加以下配置：
|
||||
# 消息队列类型
|
||||
COZE_MQ_TYPE=pulsar
|
||||
|
||||
# Pulsar 服务地址
|
||||
MQ_NAME_SERVER=pulsar://pulsar:6650
|
||||
|
||||
# JWT 认证 Token（可选，如果启用了认证）
|
||||
# PULSAR_JWT_TOKEN=your_jwt_token_here
|
||||
```
|
||||
|
||||
### 4. 启动服务
|
||||
|
||||
```bash
|
||||
# 启动包含 Pulsar 的完整 Coze Studio 服务
|
||||
docker-compose up -d
|
||||
|
||||
# 查看服务启动状态
|
||||
docker-compose ps
|
||||
```
|
||||
|
||||
### 5. 验证部署
|
||||
|
||||
```bash
|
||||
# 检查 Pulsar 容器状态
|
||||
docker ps | grep pulsar
|
||||
|
||||
# 检查 Pulsar 健康状态
|
||||
curl -f http://localhost:8080/admin/v2/clusters
|
||||
|
||||
# 查看 Pulsar 日志
|
||||
docker logs coze-pulsar
|
||||
|
||||
# 测试 Pulsar 连接
|
||||
docker exec -it coze-pulsar bin/pulsar-admin clusters list
|
||||
```
|
||||
|
||||
### 6. 访问服务
|
||||
|
||||
- **Coze Studio**: `http://localhost:3000`（根据实际配置）
|
||||
- **Pulsar Admin**: `http://localhost:8080`
|
||||
|
||||
现在 Coze Studio 已经成功集成 Pulsar 作为消息队列，所有的事件总线功能都将通过 Pulsar 处理。
|
||||
|
||||
## 附录
|
||||
|
||||
### A. 生产环境集群部署
|
||||
|
||||
生产环境推荐使用 Pulsar 集群部署以获得高可用性和更好的性能。集群部署涉及 ZooKeeper、BookKeeper 和 Broker 等多个组件的配置，配置较为复杂。
|
||||
|
||||
**生产环境建议**：
|
||||
- 使用 Pulsar 集群模式部署，确保高可用性
|
||||
- 开启 JWT 认证，保障系统安全
|
||||
- 配置适当的资源限制和监控
|
||||
|
||||
详细的集群部署配置请参考 [Apache Pulsar 官方文档](https://pulsar.apache.org/docs/4.1.x/deploy-kubernetes/)。
|
||||
|
||||
### B. 可视化管理工具
|
||||
|
||||
对于需要图形化界面管理 Pulsar 集群的用户，可以考虑使用 ASP 社区版。ASP 社区版是一个专为 Apache Pulsar 设计的现代化管理平台，提供了直观的 Web 界面来管理集群、租户、命名空间、主题等资源。该平台支持实时监控、性能指标展示、配置管理等功能，大大简化了 Pulsar 集群的日常运维工作。
|
||||
|
||||
更多信息请参考：[ASP 社区版文档](https://ascentstream.com/docs/asp/asp-community/overview)
|
||||
|
||||
### C. 适配特点
|
||||
|
||||
#### 1. 设计原则
|
||||
|
||||
**架构兼容性设计**：
|
||||
- 严格遵循 Coze Studio EventBus 接口规范，确保与现有系统无缝集成
|
||||
- 采用工厂模式实现多种 MQ 的统一管理
|
||||
- 保持与 NSQ、Kafka、RocketMQ 实现的接口一致性
|
||||
|
||||
**性能优先**：
|
||||
- 异步批量发送减少网络开销
|
||||
- 连接池复用降低连接成本
|
||||
- 消息确认机制保证可靠性
|
||||
|
||||
**易于部署**：
|
||||
- 单机模式快速启动
|
||||
- Docker 容器化部署
|
||||
- 环境变量配置，灵活易用
|
||||
|
||||
#### 2. 技术亮点
|
||||
|
||||
**JWT 认证支持**：
|
||||
```go
|
||||
// 自动检测和配置 JWT 认证
|
||||
if jwtToken := os.Getenv(consts.PulsarJWTToken); jwtToken != "" {
|
||||
clientOptions.Authentication = pulsar.NewAuthenticationToken(jwtToken)
|
||||
logs.Debugf("Using JWT authentication, token length: %d", len(jwtToken))
|
||||
}
|
||||
```
|
||||
|
||||
**批量发送优化**：
|
||||
```go
|
||||
// 异步批量发送提高性能
|
||||
for _, body := range bodyArr {
|
||||
msg := &pulsar.ProducerMessage{Payload: body}
|
||||
if option.ShardingKey != nil {
|
||||
msg.Key = *option.ShardingKey
|
||||
}
|
||||
p.producer.SendAsync(ctx, msg, callback)
|
||||
}
|
||||
```
|
||||
|
||||
**优雅关闭处理**：
|
||||
```go
|
||||
// 监听系统信号，优雅关闭资源
|
||||
safego.Go(context.Background(), func() {
|
||||
signal.WaitExit()
|
||||
logs.Infof("shutting down pulsar consumer for topic: %s, group: %s", topic, group)
|
||||
cancel()
|
||||
consumer.Close()
|
||||
client.Close()
|
||||
})
|
||||
```
|
||||
|
||||
### C. 故障排查
|
||||
|
||||
#### 1. 常见问题
|
||||
|
||||
**连接问题**：
|
||||
```bash
|
||||
# 检查 Pulsar 服务状态
|
||||
docker exec -it coze-pulsar bin/pulsar-admin brokers healthcheck
|
||||
|
||||
# 检查网络连通性
|
||||
telnet localhost 6650
|
||||
|
||||
# 查看连接配置
|
||||
docker exec -it coze-pulsar cat conf/standalone.conf | grep -E "(advertisedAddress|bindAddress)"
|
||||
```
|
||||
|
||||
**认证问题**：
|
||||
```bash
|
||||
# 检查 JWT Token 配置
|
||||
echo $PULSAR_JWT_TOKEN
|
||||
|
||||
# 验证 Token 有效性
|
||||
docker exec -it coze-pulsar bin/pulsar-admin --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
|
||||
--auth-params token:$PULSAR_JWT_TOKEN \
|
||||
clusters list
|
||||
```
|
||||
|
||||
**性能问题**：
|
||||
```bash
|
||||
# 查看 Topic 积压情况
|
||||
docker exec -it coze-pulsar bin/pulsar-admin topics stats persistent://public/default/your-topic
|
||||
|
||||
# 调整批量发送参数
|
||||
# 在代码中可以通过 SendOpt 配置批量大小和延迟
|
||||
```
|
||||
|
||||
#### 2. 日志分析
|
||||
|
||||
```bash
|
||||
# 查看 Pulsar 服务日志
|
||||
docker logs coze-pulsar
|
||||
|
||||
# 查看应用日志中的 Pulsar 相关信息
|
||||
tail -f logs/coze-studio.log | grep -i "pulsar\|eventbus"
|
||||
|
||||
# 启用详细日志
|
||||
# 在 Pulsar 配置中设置 rootLogLevel=DEBUG
|
||||
```
|
||||
|
||||
#### 3. 监控指标
|
||||
|
||||
```bash
|
||||
# 获取 Broker 指标
|
||||
curl http://localhost:8080/metrics/
|
||||
|
||||
# 获取特定 Topic 指标
|
||||
curl http://localhost:8080/admin/v2/persistent/public/default/your-topic/stats
|
||||
|
||||
# 监控消费延迟
|
||||
docker exec -it coze-pulsar bin/pulsar-admin topics subscriptions persistent://public/default/your-topic
|
||||
```
|
||||
|
||||
### D. 最佳实践
|
||||
|
||||
#### 1. 生产环境配置
|
||||
|
||||
```bash
|
||||
# 推荐的生产环境配置
|
||||
COZE_MQ_TYPE=pulsar
|
||||
MQ_NAME_SERVER=pulsar://pulsar-broker-1:6650,pulsar://pulsar-broker-2:6650,pulsar://pulsar-broker-3:6650
|
||||
PULSAR_JWT_TOKEN=your-production-jwt-token
|
||||
|
||||
# Pulsar 集群配置
|
||||
# 建议至少 3 个 Broker 节点
|
||||
# 建议至少 3 个BookKeeper 节点
|
||||
# 建议至少 3 个 ZooKeeper 节点
|
||||
```
|
||||
|
||||
#### 2. 性能调优
|
||||
|
||||
```bash
|
||||
# Producer 配置优化
|
||||
# 批量发送大小：1000 条消息或 1MB
|
||||
# 发送超时：30 秒
|
||||
# 压缩算法：LZ4
|
||||
|
||||
# Consumer 配置优化
|
||||
# 接收队列大小：1000
|
||||
# 确认超时：30 秒
|
||||
# 消费者类型：Exclusive（保证顺序）
|
||||
```
|
||||
|
||||
#### 3. 安全配置
|
||||
|
||||
```bash
|
||||
# 启用 JWT 认证
|
||||
PULSAR_JWT_TOKEN=your-jwt-token
|
||||
|
||||
# 配置访问控制列表（ACL）
|
||||
# 通过 Pulsar Admin 工具配置 Topic 级别的权限
|
||||
```
|
||||
|
||||
## 总结
|
||||
|
||||
Apache Pulsar 在 Coze Studio 中的 EventBus 集成实现了以下目标：
|
||||
|
||||
1. **高性能**: 支持高吞吐量、低延迟的消息传递
|
||||
2. **高可靠**: 消息持久化存储，支持消息确认机制
|
||||
3. **易扩展**: 支持水平扩展，适应业务增长
|
||||
4. **易运维**: 丰富的管理工具和监控指标
|
||||
5. **企业级**: 多租户支持，适合企业级应用场景
|
||||
|
||||
通过这次集成，Coze Studio 为用户提供了一个高性能、高可靠、易扩展的消息队列解决方案，特别适合需要高吞吐量、低延迟、企业级特性的场景。
|
||||
|
||||
## 相关链接
|
||||
|
||||
- [Apache Pulsar 官方文档](https://pulsar.apache.org/docs/)
|
||||
- [Pulsar Go Client 文档](https://pulsar.apache.org/docs/client-libraries-go/)
|
||||
- [ASP 社区版文档](https://ascentstream.com/docs/asp/asp-community/overview)
|
||||
- [Coze Studio 项目地址](https://github.com/coze-dev/coze-studio)
|
||||
Reference in New Issue
Block a user