mossrose

轻量分布式作业框架

---

特性

• 通过选举保证集群中只出现一个主节点，负责trigger作业
• 所有集群节点参与作业计算任务
• 内置多种作业类型，并预留了SPI接口支持扩展

文档

• Wiki: https://github.com/jiuxiantuan/mossrose/wiki
• Example: https://github.com/jiuxiantuan/mossrose-example
• Spring Boot Starter: https://github.com/jiuxiantuan/mossrose-spring-boot-starter
• Spring Boot Example: https://github.com/jiuxiantuan/mossrose-spring-boot-example

Requirement

• Zookeeper
• Java 8
• Spring 3.x+

Installation

<dependency>
  <groupId>com.jiuxian</groupId>
  <artifactId>mossrose</artifactId>
  <version>2.4.4-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);
			}
		};
	}

}