项目背景
最近在做公司项目的微服务改造，在dubbo和spring-cloud这两个主流的微服务框架之前技术选型徘徊了好久，两个框架在性能，设计，社区支持都非常地完美。然而我为什么要选择自己去封装ignite去实现RPC服务呢？
- 原因一：
公司缺乏专门的运维人员去维护第三方的中间件服务，例如 Redis，Zookeeper，Rabbitmq等等的这些中间件。 - 原因二：
应用本身要求的高并发不多，但是需要做高可用，引入dubbo和spring-cloud视乎有点小题大做的感觉 - 原因三：
使用dubbo和spring-cloud对现有项目的代码改造工作量有点大，本身公司项目有部分已经基于ignite做服务网格了 - 原因四：
工作这么久了，还是要尝试封装一下框架，所以这个框架是我的处女做，希望大家能够批评和指正
项目概述
底层基于apache ignite，特点是可以做到不依赖外部中间件，实现RPC服务，分布式缓存，分布式计算，分布式消息等功能特性
框架也基于ignite的集群管理，实现了基于集群组的颗粒度的服务调用，即针对集群组的调用
- JDK和Spring boot版本
JDK版本为1.8
Spring boot 版本要求1.5.3以上
框架说明
服务中启动的 Spring boot 应用同时启动了ignite的server和client模式，注入到了Spring容器中
@Autowired
@Qualifier("igniteClient")
private Ignite igniteClient;
@Autowired
@Qualifier("igniteServer")
private Ignite igniteServer;
因此你可以无缝地使用框架没有封装的ignite功能，更多的ignite的功能，请参考中文官网
（https://www.ignite-service.cn/doc/java/）
功能概述
- RPC服务
- 分布式消息
- 分布式广播
- 分布式计算
quick-start
构建
git clone https://github.com/konglinghai123/ignite-spring-project.git
cd ignite-spring-boot-starter
mvn clean install
构建一个基于ignite的 spring boot 项目
- 添加依赖:
<dependency>
<groupId>com.github.kong.spring.boot</groupId>
<artifactId>ignite-spring-boot-starter</artifactId>
<version>1.0</version>
</dependency>
在application-yml添加ignite的相关配置信息，样例配置如下:
zookeeper 发现
#zookeeper发现
ignite-cluster:
name: hello_client_1 #节点名称
role: client
des: 测试服务
zookeeperUrl: 192.168.56.100:2181
localAddress: 127.0.0.1
localPort: 47600
- 动态ip发现
ignite-cluster:
name: hello_server #节点名称
role: server
des: 测试服务端
multicast-group: 224.0.1.111 #组播地址
localAddress: 127.0.0.1
localPort: 48600
- 为了开发方便，如果Spring boot Appliction 类的不在包名
com.github.kong目录下，接下来在Spring Boot Application的上添加@ComponentScan("com.github.kong.*")，这样idea可以通过看到一些Bean是否已经注入了，当然也可以不添加，框架也有写扫描注入
@SpringBootApplication
@ComponentScan("com.github.kong.*")
public class HelloWorldServerApplication {
public static void main(String[] args) {
SpringApplication.run(HelloWorldServerApplication.class);
}
}
RPC服务的创建与消费
发布服务基于ignite的RPC服务
- 编写你的ignite服务，需要添加要发布的服务实现上添加
@IgniteRpcService注解，继承IgniteService. -
HelloWorld是定义的接口
@Service
@IgniteRpcService(des = "这是一个例子")
public class HelloWorldService extends IgniteService implements HelloWorld {
}
-
@IgniteRpcService注解的定义
/**
* 服务提供者注解
*/
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface IgniteRpcService {
/**
* @return
*/
String version() default "1.0";
//接口描述
String des() default "";
//单个节点部署的实例数
int maxPerNodeCount() default 1;
//整个集群部署的最大实例数，0：无限制
int total() default 0;
}
- 启动你的Spring Boot应用，观察控制台，可以看到ignite启动相关信息.
调用已经发布的RPC服务
- Spring boot 应用配置同上，唯一不同的是，需要更改配置
#zookeeper发现
ignite-cluster:
name: hello_client_1 #节点名称 (必须在集群中唯一)
- 通过
@IgniteRpcReference注入需要使用的interface.
@Controller
public class HelloWorldController {
@IgniteRpcReference
private HelloWorld helloWorldService;
@RequestMapping("/helloworld")
@ResponseBody
public String test(){
return helloWorldService.sayHello("kong");
}
}
- 调用不同版本的RPC服务
@IgniteRpcReference(version = "1.1")
private HelloWorld helloWorldService;
-
@IgniteRpcReference注解的定义
/**
* 网格服务注入注解
*/
@Target({ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface IgniteRpcReference {
String version() default "1.0";
//默认使用负载均衡
boolean isLoadbalance() default true;
//默认不设超时
long timeout() default 0;
}
分布式消息
分布式消息是基于内存的消息订阅系统，如果需要持久化请使用外部的消息系统
定义话题消费者
@Service
@IgniteMessageListener(topic = "hello",isBroadcast = false)
public class HelloWorldMessage implements IgniteMessageRecevicer<String> {
@Override
public boolean apply(UUID uuid, MessageModel<String> messageModel) {
System.out.println(messageModel);
return true;
}
}
-
@IgniteMessageListener注解的定义
/**
* 服务提供者注解
*/
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface IgniteMessageListener {
//消息主题
String topic();
//消息描述
String des() default "";
//是否针对集群内的所有节点（是否允许重复消费）
boolean isBroadcast() default true;
}
-
MessageModel是一个消息封装，发送消息时必须用它来发送
发送话题消息
@Controller
@RequestMapping("/message")
public class MessageController {
@Autowired
private IgniteMessageSender sender;
@RequestMapping("/sayHello")
@ResponseBody
public String test(){
sender.toRemote("hello", new MessageModel<>("1212"));
return "1212";
}
}
-
IgniteMessageSender是框架注入的Bean，可以直接引用
<span id="BroadCast">分布式广播</span>
分布式广播是指：对集群组的所有节点发送消息，然后获取所有节点返回的结果，原来是基于ignite的分布式闭包利用反射机制调用spring容器内Bean的方法
发送一个分布式广播
@Controller
@RequestMapping("/broadcast")
public class BroadcastController {
@Autowired
private BroadcastServiceExecutor broadcastServiceExecutor;
@RequestMapping("/sayHello")
@ResponseBody
public List<String> test(){
return (List<String>) broadcastServiceExecutor.broadcast("server", TestBroadService.class,"sayHello","12123");
}
}
BroadcastServiceExecutor是框架注入的Bean，可以直接引用broadcast方法提供3个方法定义
/**
* 向其他集群广播
* @param targetRole 集群标识
* @param targetClass api类
* @param methodName 方法名称
* @param args 参数
* @return
*/
public List broadcast(String targetRole,Class targetClass,String methodName,Object... args){...}
/**
* 向远端集群广播消息
* @param targetClass
* @param methodName
* @param args
* @return
*/
public List broadcastRemote(Class targetClass,String methodName,Object... args){...}
/**
* 向集群内广播消息
* @param targetClass
* @param methodName
* @param args
* @return
*/
public List broadcastLocal(Class targetClass,String methodName,Object... args){...}
分布式计算
分布式计算允许用户执行基于内存的Map-Reduce任务
- 创建
Map-Reduce任务 ,需继承ComputeTaskSplitAdapter(import org.apache.ignite.compute.ComputeTaskSplitAdapter)，泛型<T,R> - T:入参，R：返回类型
//字数统计测试
@Service
public class MapExampleCharacterCountTask extends ComputeTaskSplitAdapter<List<String>,Integer> {
@Nullable
@Override
public Integer reduce(List<ComputeJobResult> results) throws IgniteException {
return results.stream().mapToInt(ComputeJobResult::<Integer>getData).sum();
}
@Override
protected Collection<? extends ComputeJob> split(int gridSize, List<String> arg) throws IgniteException {
LinkedList jobs = new LinkedList();
List<List<String>> list = CollectionUtils.split(arg,10000);
for(final List<String> words : list){
jobs.add(new ComputeJobAdapter() {
@Override
public Object execute() throws IgniteException {
int i = 0;
for(String s : words){
i = i + s.length();
}
return i;
}
});
}
return jobs;
}
}
- 执行
Map-Reduce任务
@Controller
@RequestMapping("/mr")
public class MRTestController {
@Autowired
private MapReduceTaskExecutor<List<String>,Integer> mapReduceTaskExecutor;
@RequestMapping("/test")
@ResponseBody
public Object test(){
try {
List<String> records = new ArrayList<>();
// 创建CSV读对象
CsvReader csvReader = new CsvReader(new FileInputStream("D:\\data\\cs2.csv"), Charset.forName("GBK"));
while (csvReader.readRecord()){
// 读一整行
records.add(csvReader.getRawRecord());
}
List<String> bigRecords = new ArrayList<>();
for(int i = 0; i < 5; i++){
bigRecords.addAll(records);
}
return mapReduceTaskExecutor.execute(MapExampleCharacterCountTask.class,bigRecords);
} catch (IOException e) {
e.printStackTrace();
}
return "";
}
}
-
MapReduceTaskExecutor是框架注入的Bean，可以直接引用
集群管理api
框架注入了IgniteManager这个bean，可以实现以下功能
public interface IgniteManager {
/**
* 获取节点列表
*
* @return
*/
List<NodeInfo> list();
/**
* 获取节点的详细信息
*
* @param nodeId
* @return
*/
ClusterMetrics info(String nodeId);
/**
* 获取微服务的基本信息
*
* @return
*/
List<ServiceInfo> servieInfos();
/**
* 集群消息信息
* @return
*/
List<MessageInfo> messagInfos();
}
- 使用
@Autowired注入即可
@Controller
@RequestMapping("/admin")
public class AdminCotroller {
@Autowired
private IgniteManager igniteManager;
@RequestMapping("/nodes")
@ResponseBody
public List<NodeInfo> nodes(){
return igniteManager.list();
}
@RequestMapping("/nodeInfo/{id}")
@ResponseBody
public ClusterMetrics info(@PathVariable("id") String id){
return igniteManager.info(id);
}
@RequestMapping("/services")
@ResponseBody
public List<ServiceInfo> services(){
return igniteManager.servieInfos();
}
}
