Gearman 基础 以及 Gearman 使用举例

<一> Gearman 基础

Gearman 是什么？

Gearman 是一个分布式的任务分发框架。

Gearman 用来把请求委派给机器，提供通用的程序框架来将任务分发在机器运算。同时具备并行工作能力、负载均衡处理的能力，以及跨语言通信能力。

Client 的工作原理

主要分成三个部分Client、Job、Worker：

client：负责建立一个工作，发送请求给Job Server，而Job Server 会去找合适的 Worker 去转发工作。

Job Server：了解Client 端的请求，并查看哪个机器可以处理这项请求，在系统里它通常是个Daemon。

Worker：Worker 通过Job Server 的分派，开始执行Client 端的工作。

关于Message Queue

执行 Message Queue 服务的 Job Server 可以是多台服务器组成，也就是分布式架构，在 Job Server 上执行 Worker 程序。

这些 Worker 程序会一直循环地等候，直到 Job Server 呼叫它执行工作。

Client 端发送出请求之后，会将需要的资料及动作记录在 Job Server 上，这时 Job Server 会查看是否有空闲并符合需求的 Worker。

在 Worker 结束工作后，会发送通知给 Job Server ，这时 Job Server 就会视状况把结果回传给 Client。

Client 端不需等候需求的执行结果，可以直接继续执行其他动作。

Job Server 负载方式

当 Client 可能同时发出多个请求给 Job Server，由 Message Queue 接手进行处理。

而 Job Server 开始处理多个请求，若其中一个发生问题，可以 Failover 到其他的机器。

同时，Worker 会将多个请求一起进行运算，再看是同步或异步模式，回传结果给 Client。

同步 (Synchronous)

同步(Synchronous) 是指 Client 将请求 (Application) 丢给 Gearmand。

由 Gearmand 分派 Job 给各 Worker 去处理。

并同步 Response 回传给 Gearmand 告诉 Client 现在进度。

异步 (Asynchronous)

异步 (Asynchronous) 是指 Client 将请求 (Application) 丢给 Gearmand。

由 Gearmand 分派 Job 给各 Worker 去处理。

Worker 处理完毕后，才会将结果回传给 Gearmand 告诉 Client 现在进度。

<二> Gearman 举例

来自Google Code 的例子，并做了一些修改 和 添加了一些注释。

client 提交work 到 server，server 将任务分配给worker。

worker 所做得是将待处理的数据反转。

1.Gearman server：

public class EchoWorkerServer {
public static void main(String... args) throws IOException {
// 创建一个Gearman 实例
Gearman gearman = Gearman.createGearman();

try {
// 启动一个新的 job server.
// 在本机启动
// 参数为监听端口
GearmanServer server = gearman.startGearmanServer(EchoWorker.ECHO_PORT);

// 创建一个 gearman worker.
// 从server 拉任务，然后执行相应的 GearmanFunction
GearmanWorker worker = gearman.createGearmanWorker();

// 告诉worker 如何执行工作(主要实现了 GearmanFunction 接口)
worker.addFunction(EchoWorker.ECHO_FUNCTION_NAME, new EchoWorker());

// worker 连接服务器
worker.addServer(server);

} catch (IOException ioe) {
// 出现错误，关闭 gearman service
gearman.shutdown();
throw ioe;
}
}
}

2.Gearman worker

public class EchoWorker implements GearmanFunction {
// function 名称
public static final String ECHO_FUNCTION_NAME = "echo";

// job server 的地址
// Echo host 为安装了Gearman 并开启Gearman 服务的主机地址
public static final String ECHO_HOST = "localhost";

// job server监听的端口  默认的端口
public static final int ECHO_PORT = 4730;

public static void main(String... args) {

// 创建一个Gearman 实例
Gearman gearman = Gearman.createGearman();

// 创建一个job server
// 参数1：job server 的地址
// 参数2：job server 监听的端口
// job server 收到 client 的job，并将其分发给注册的worker
GearmanServer server=gearman.createGearmanServer(EchoWorker.ECHO_HOST,EchoWorker.ECHO_PORT);

// 创建一个 Gearman worker
GearmanWorker worker = gearman.createGearmanWorker();

// 告诉 worker 如何执行工作
worker.addFunction(EchoWorker.ECHO_FUNCTION_NAME, new EchoWorker());

// worker 连接服务器
worker.addServer(server);
}

@Override
public byte[] work(String function, byte[] data, GearmanFunctionCallback callback) throws Exception {
// work方法实现了GearmanFunction接口中的work方法,本实例中进行了字符串的反写
if (data != null) {
String str = new String(data);
StringBuffer sb = new StringBuffer(str);
return sb.reverse().toString().getBytes();
} else {
return "未接收到data".getBytes();
}
}
}

3.Gearman Client

public class EchoClient {
public static void main(String... args) throws InterruptedException {
Gearman gearman = Gearman.createGearman();

// 创建一个 Gearman client
// 用来向 job server 提交请求
GearmanClient client = gearman.createGearmanClient();

// 创建一个 job server 对象，该对象代表 remote job server.
// job server 从clients 得到jobs 然后分发给注册workers
GearmanServer server = gearman.createGearmanServer(
EchoWorker.ECHO_HOST, EchoWorker.ECHO_PORT);

// client 连接server
client.addServer(server);

// 将 job submit 给server
// 参数1：function 名称
// 参数2：将要传给server 和 worker 的数据
// GearmanJobReturn 用来取得 job 的结果
GearmanJobReturn jobReturn = client.submitJob(
EchoWorker.ECHO_FUNCTION_NAME, ("Hello World").getBytes());

// 遍历job 事件，直到文件末尾
while (!jobReturn.isEOF()) {
// 获得下个job，(阻塞操作)
GearmanJobEvent event = jobReturn.poll();
switch (event.getEventType()) {
// success
case GEARMAN_JOB_SUCCESS: // Job completed successfully
// print the result
System.out.println(new String(event.getData()));
break;

// failure
case GEARMAN_SUBMIT_FAIL: // The job submit operation failed
case GEARMAN_JOB_FAIL: // The job's execution failed
System.err.println(event.getEventType() + ": "
+ new String(event.getData()));
}
}
gearman.shutdown();
}
}

说明：server 和 worker 都是在local host 运行