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 运行