processor

Documentation

Overview

Package processor 提供 Python 脚本 Processor 实现。 通过子进程启动 Python 脚本，从 Go channel 读取数据通过 stdin 发送给 Python， Python 处理后从 stdout 读取结果发送到下游 Go channel。

Index

• type Config
• type Processor
  • func New() *Processor
  • func (p *Processor) Close() error
  • func (p *Processor) ConcurrencyCap() dataflow.ConcurrencyCap
  • func (p *Processor) Init(config []byte) error
  • func (p *Processor) Process(ctx context.Context, in <-chan types.Record, out chan<- types.Record) error

Types

type Config

type Config struct {
	// Script Python 脚本路径（必填）
	Script string `json:"script"`
	// PythonExec Python 可执行文件路径，默认 "python3"
	PythonExec string `json:"python_exec"`
	// Args 传递给脚本的额外命令行参数
	Args []string `json:"args"`
	// Env 额外的环境变量（值支持 string/number/bool，自动转为字符串）
	Env map[string]interface{} `json:"env"`
}

Config Processor 的配置。

type Processor

type Processor struct {
	// contains filtered or unexported fields
}

Processor 通过 Python 脚本处理数据的 Processor。 使用双向通信：Go 写 stdin → Python 处理 → Go 读 stdout。 采用流式模式：stdin 写入和 stdout 读取并发进行，利用管道自然背压保证不丢数据。

func New

func New() *Processor

New 创建 Processor 实例。

func (*Processor) Close

func (p *Processor) Close() error

Close 关闭 Python 进程。

func (*Processor) ConcurrencyCap

func (p *Processor) ConcurrencyCap() dataflow.ConcurrencyCap

ConcurrencyCap 返回并发能力（不支持并发，因为有状态的子进程通信）。

func (*Processor) Init

func (p *Processor) Init(config []byte) error

Init 解析配置。

func (*Processor) Process

func (p *Processor) Process(ctx context.Context, in <-chan types.Record, out chan<- types.Record) error

Process 从输入通道读取数据，发送给 Python 处理，将结果发送到输出通道。 采用流式模式：stdin 写入和 stdout 读取并发进行，利用管道自然背压保证不丢数据。