benthos 是一个stream 处理框架，streamsets 也是，但是两者可以通过不同的工具进行集成起来

一般我们可以使用http 服务，消息中间件（kafka、rabbitmq 。。。）

使用docker-compose 运行

服务配置

• docker-compose 文件

version: "3"services:  sets:    image: streamsets/datacollector     volumes:     - "./ms/data:/data"     - "./ms/logs:/logs"     - "./ms/tmp:/tmp"    ports:     - "8000:8000"     - "18630:18630"  redis:    image: redis    ports:     - "6379:6379"  benthos-in:    image: jeffail/benthos    environment:      INPUT_TYPE: "http_server"      INPUT_HTTP_SERVER_ADDRESS: "0.0.0.0:8080"      OUTPUT_TYPE: "amqp"      OUTPUT_AMQP_URL: "amqp://guest:guest@rabbitmq:5672/appdemo"      OUTPUT_AMQP_EXCHANGE: "benthos-exchange"      OUTPUT_AMQP_EXCHANGE_TYPE: "direct"      OUTPUT_AMQP_KEY: "benthos-key"    ports:      - "8080:8080"  rabbitmq:    image: rabbitmq:3.6.14-management    ports:      - "4369:4369"      - "5671:5671"      - "5672:5672"      - "25672:25672"      - "15672:15672"

• 启动

docker-compose up -d

• 添加streamsets rabbitmq 包
  
• rabbitmq vhost 创建
  

说明

benthos 使用http service 与rabbitmq 集成的模式，streamsets 使用rabbitmq customer 与local fs 集成的模式

benthos 服务的启动 curl http://localhost:8080/post -d "example message"

同时首选需要进行vhost 的创建，目前发现的一个bug的使用默认的就会有问题，需要单独创建，可以使用管理界面

http://docker-host:15672

streamsets 配置

• pipeline flow
  
• rabbitmq 配置
  

• local fs
  
• benthos 配置说明

参考docker-compose 文件environment:      INPUT_TYPE: "http_server"      INPUT_HTTP_SERVER_ADDRESS: "0.0.0.0:8080"      OUTPUT_TYPE: "amqp"      OUTPUT_AMQP_URL: "amqp://guest:guest@rabbitmq:5672/appdemo"      OUTPUT_AMQP_EXCHANGE: "benthos-exchange"      OUTPUT_AMQP_EXCHANGE_TYPE: "direct"      OUTPUT_AMQP_KEY: "benthos-key"

启动pipeline

效果

• 访问(发送消息)

curl http://localhost:8080/post -d "rongfengliang@qq.com"

• 效果
  
  

配置问题说明

目前发现streamsets rabbit 配置必须包含一个vhost 不能使用默认的。

参考资料