使用mossrose构建分布式作业集群

使用mossrose构建分布式作业集群

社区

Yahoo Group: https://groups.yahoo.com/group/mossrose

QQ群：595011342

文档

Wiki: https://github.com/jiuxiantuan/mossrose/wiki

Requirement

Zookeeper

Java 8

Spring 3.x+

非Spring用户：[]https://github.com/jiuxiantuan/mossrose/wiki/Use-mossrose-without-spring]

Installation

<dependency>
<groupId>com.jiuxian</groupId>
<artifactId>mossrose</artifactId>
<version>2.2.0-RELEASE</version>
</dependency>

Key concept

SimpleJob

简单任务

DistributedJob

分布式任务，通过Slicer将作业分隔成多个子任务，子任务在集群内分布执行

StreamingJob

分布式流式任务，解决分片非常多时DistributedJob内存占用大的问题

MapReduceJob

MapReduce任务

MossroseProcess

多个MossroseProcess组成集群，集群保证有且只有一个节点竞选成为主节点，主节点负责触发作业；所有节点都是工作节点，主节点触发的任务会在所有工作节点上分布执行

MossroseConfig

Mossrose配置，包括集群元信息和任务元信息

Quick Start

Implement a simple job

public class SimpleExampleJob implements SimpleJob {

@Override
public Executor executor() {
return new Executor() {

@Override
public void execute() {
LOGGER.info("SimpleJob");
}
};
}

}

Config the job in spring

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:mossrose="https://jiuxiantuan.github.io/mossrose"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd https://jiuxiantuan.github.io/mossrose https://jiuxiantuan.github.io/mossrose/mossrose.xsd"> 
<bean class="com.jiuxian.jobs.bean.BusinessBean" />
<bean id="simpleExampleJob" class="com.jiuxian.jobs.job.SimpleExampleJob" />
<bean id="distributedExampleJob" class="com.jiuxian.jobs.job.DistributedExampleJob" />
<bean id="streamingExampleJob" class="com.jiuxian.jobs.job.StreamingExampleJob" />

<mossrose:springholder />
<mossrose:config>
<mossrose:cluster name="mossrose-example" discovery-zk="localhost:2181" />
<mossrose:jobs>
<mossrose:job id="SimpleExampleJob" cron="0/5 * * * * ?" job-bean-name="simpleExampleJob" group="example" />
<mossrose:job id="DistributedExampleJob" cron="0/15 * * * * ?" job-bean-name="distributedExampleJob" group="example" />
<mossrose:job id="StreamingExampleJob" cron="0/20 * * * * ?" job-bean-name="streamingExampleJob" group="example"
description="分布式流式任务示例" />
</mossrose:jobs>
</mossrose:config>
<mossrose:process />
<mossrose:ui />

</beans>

Start the job

applicationContext.getBean(MossroseProcess.class).run();

Distributed Job

Implement a distributed job

public class SomeDistributedJob implements DistributedJob<String> {

private static final Logger LOGGER = LoggerFactory.getLogger(SomeDistributedJob.class);

@Override
public Slicer<String> slicer() {
return new Slicer<String>() {

@Override
public List<String> slice() {
return Splitter.on(" ").splitToList("This is a test on the mossrose distributed job, how are you feeling?");
}
};
}

@Override
public com.jiuxian.mossrose.job.DistributedJob.Executor<String> executor() {
return new Executor<String>() {

@Override
public void execute(String item) {
LOGGER.info(Thread.currentThread() + " DistributedJob: " + item);
}
};
}

}

Streaming Job

Implement a streaming job

DistributedJob需要把需要分布式执行的任务集合一次性的返回，在集合非常大的时候会存在内存的问题，StreamingJob解决了这个问题，任务可以以流的方式不断输出，以保证内存可以及时释放。

public class SomeStreamingJob implements StreamingJob<String> {

private static final Logger LOGGER = LoggerFactory.getLogger(SomeStreamingJob.class);

@Override
public Streamer<String> streamer() {
return new Streamer<String>() {

private List<String> list = Lists.newArrayList("Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday");

private int index = 0;

@Override
public boolean hasNext() {
return index < list.size();
}

@Override
public String next() {
return list.get(index++);
}
};
}

@Override
public Executor<String> executor() {
return new Executor<String>() {

@Override
public void execute(String item) {
LOGGER.info(Thread.currentThread() + " StreamingJob: " + item);
}
};
}

}

MapReduce Job

Implement a map/reduce job

public class MapReduceExampleJob implements MapReduceJob<Integer, Integer> {

@Override
public com.jiuxian.mossrose.job.MapReduceJob.Mapper<Integer> mapper() {
return new Mapper<Integer>() {

@Override
public List<Integer> map() {
return Lists.newArrayList(1, 2, 3, 4, 5, 6, 7);
}
};
}

@Override
public com.jiuxian.mossrose.job.MapReduceJob.Executor<Integer, Integer> executor() {
return new Executor<Integer, Integer>() {

@Override
public Integer execute(Integer item) {
return item * 2;
}
};
}

@Override
public com.jiuxian.mossrose.job.MapReduceJob.Reducer<Integer> reducer() {
return new Reducer<Integer>() {

@Override
public void reduce(List<Integer> rs) {
LOGGER.info("Reduce result : {}", rs);
}
};
}

}