R3 FRAMEWORK - 分布式实时处理框架

    public interface SayHelloService {
        @Out(SayHelloOutHandler.class)
        @In(SayHelloInHandler.class)
        List<String> sayHello(List<String> names);
    }

    public class SayHelloOutHandler extends OutHandler {
        @Override
        public List<Object[]> shard(Object[] parameters, int shardCount) throws Throwable {
            List<Object[]> result = new ArrayList<>();
            List<String> param = (List<String>) parameters[0];
            int perSize = param.size() / shardCount;
            for(int i=0;i<shardCount;i++){
                result.add(new Object[]{param.subList(i*perSize,(i!=shardCount-1?(i+1)*perSize:param.size()))});
            }
            return result;
        }
    }

    public class SayHelloInHandler extends InHandler {
        @Override
        public Object join(Object[] results) throws Throwable {
            List<String> list = new ArrayList<>();
            for(Object object : results){
                list.addAll((List<String>)object);
            }
            return list;
        }
    }

public class SayHelloServiceImpl implements SayHelloService {
    @Override
    public List<String> sayHello(List<String> names) {
        List<String> newNames = new ArrayList<>();
        for (String name : names) {
            String newName = "hello " + name;
            newNames.add(newName);
        }
        return newNames;
    }
}

   <r3:application name="worker-1" host="AUTO" port="20080"></r3:application> 
   <r3:registry address="localhost:2181" namespace="r3"></r3:registry> 
   <r3:worker id="sayHelloWorker" ref="sayHelloServiceImpl" interface="r3.example.api.SayHelloService"></r3:worker> 
   <bean id="sayHelloServiceImpl" class="r3.example.worker.SayHelloServiceImpl" scope="prototype"></bean>

   <r3:application name="leader-one" ></r3:application>
   <r3:registry address="localhost:2181" namespace="r3"></r3:registry> 
   <r3:leader id="sayHelloServiceImpl" interface="r3.example.api.SayHelloService"></r3:leader>