PHPackages                             rabbit/data-pipeline - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. rabbit/data-pipeline

ActiveLibrary

rabbit/data-pipeline
====================

rabbit data pipeline

1.0.0(5y ago)06851MITPHP

Since Jun 19Pushed 1y ago1 watchersCompare

[ Source](https://github.com/rabbit-php/rabbit-data-pipeline)[ Packagist](https://packagist.org/packages/rabbit/data-pipeline)[ RSS](/packages/rabbit-data-pipeline/feed)WikiDiscussions master Synced today

READMEChangelog (1)Dependencies (4)Versions (2)Used By (1)

STS流处理
======

[](#sts流处理)

- 基于插件的流程处理组件

组件配置
----

[](#组件配置)

```
return [
        'scheduler' => [
        '{}' => Scheduler::class,
        '()' => [
            [
                '{}' => YamlParser::class,
                '()' => [
                    App::getAlias('@yaml')
                ]
            ]
        ],
        'senders' => arrdef([
            'worker' => definition(WorkerSender::class),
            'process' => definition(ProcessSender::class)
        ]),
        'cron' => definition('cron')
    ]
];
```

任务配置
----

[](#任务配置)

```
lock: task lock time,default null
cron: crontab or int,default null.int is later sec, -1 is only run once
```

通用配置
----

[](#通用配置)

```
type: Plugin Class String
start: true|false
output: Plugin Name
errHandler: callable
```

- `type`插件完整类名
- `start`是否开始插件,默认`false`
- `output`输出到下一个插件,支持`Http`输出到远程插件,`worker`多进程任务,格式
    1. `PluginName String|(http|worker):address:PluginName String`，默认不等待
    2. `['PluginName String|(http|worker):address:PluginName String':true|false|int]`

    - PS：`true`启用协程等待，不设超时;`int`启用协程等待，值为超时时间;`false`不等待
- `errHandler`错误处理函数

内置插件
----

[](#内置插件)

### Sources

[](#sources)

- [Amqp](./doc/Amqp.md)
- [FindFiles](./doc/FindFiles.md)
- [Http](./doc/Http.md)
- [Nsq](./doc/Nsq.md)
- [Pdo](./doc/pdo.md)
- [RdKafka](./doc/RdKafka.md)

---

### Common

[](#common)

- [HttpRequest](./doc/HttpRequest.md)

---

### TransForms

[](#transforms)

- [LineParser](./doc/LineParser.md)

---

### Sinks

[](#sinks)

- [Amqp](./doc/Amqp.md)
- [Clickhouse](.doc/Clickhouse.md)
- [Console](./doc/Console.md)
- [File](./doc/File.md)
- [Nsq](./doc/Nsq.md)
- [Pdo](./doc/PdoSave.md)
- [RdKafka](./doc/RdKafka.md)

### 插件开发

[](#插件开发)

- 继承`AbstractPlugin`
- 实现`run`函数
- 通过`init`从`$this->config`中获取配置参数初始化配置属性
- 插件间传值通过`Message`类，开启协程传递需要`clone Message`，输出值赋值到克隆的`Message->data`属性
- PS：配置属性禁止在运行过程中修改，插件为单例模式，协程环境下会造成属性污染。可使用局部变量替代，例如`$name=$this->name`，使用`$name`

###  Health Score

29

—

LowBetter than 60% of packages

Maintenance28

Infrequent updates — may be unmaintained

Popularity13

Limited adoption so far

Community11

Small or concentrated contributor base

Maturity55

Maturing project, gaining track record

 Bus Factor1

Top contributor holds 97.2% of commits — single point of failure

How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window.

**Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores.

**Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement.

**Maturity (30%)** — Project age, version count, PHP version support, and release stability.

###  Release Activity

Cadence

Unknown

Total

1

Last Release

2151d ago

### Community

Maintainers

![](https://www.gravatar.com/avatar/74fb4769c775c9d20ddeb66df91c19e01d1cd03785b275d51c29c953fbc03919?d=identicon)[amuluowin](/maintainers/amuluowin)

---

Top Contributors

[![amuluowin](https://avatars.githubusercontent.com/u/19545561?v=4)](https://github.com/amuluowin "amuluowin (246 commits)")[![ConlinH](https://avatars.githubusercontent.com/u/36884738?v=4)](https://github.com/ConlinH "ConlinH (7 commits)")

### Embed Badge

![Health badge](/badges/rabbit-data-pipeline/health.svg)

```
[![Health](https://phpackages.com/badges/rabbit-data-pipeline/health.svg)](https://phpackages.com/packages/rabbit-data-pipeline)
```

PHPackages © 2026

[Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)