1. 首页
  2. 文章中心
  3. java编程

使用Mina实现RPC调用，消息通知，广播。

121 阅读 0 评论
我是靠谱客的博主 动人战斗机，最近开发中收集的这篇文章主要介绍使用Mina实现RPC调用，消息通知，广播。，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

在上一篇的基础上，增强功能。
common包，定义通用接口，和传递中使用的对象，对象需要实现序列化接口。
接口:

  1. package mina.common;
  2. public interface RpcInterface {
  3.     public String getStringValue(String arg0,int arg1,Apple arg2);
  4.     public int getIntValue();
  5.     public void printPrice();
  6. }
bean类，定义了一只苹果。
  1. package mina.common;
  3. import java.awt.Color;
  4. import java.io.Serializable;
  5. import java.util.Date;
  7. public class Apple implements Serializable{
  9.     private Color color = Color.BLACK;
  10.     private double weight = 1.1;
  11.     private double dia = 2.33;
  12.     
  13.     private int num = 3;
  14.     private String name = "aaapple";
  15.     
  16.     private Date pdate = new Date();
  17.     
  18.     public Date getPdate() {
  19.         return pdate;
  20.     }
  21.     public void setPdate(Date pdate) {
  22.         this.pdate = pdate;
  23.     }
  24.     public Color getColor() {
  25.         return color;
  26.     }
  27.     public void setColor(Color color) {
  28.         this.color = color;
  29.     }
  30.     public double getDia() {
  31.         return dia;
  32.     }
  33.     public void setDia(double dia) {
  34.         this.dia = dia;
  35.     }
  36.     public String getName() {
  37.         return name;
  38.     }
  39.     public void setName(String name) {
  40.         this.name = name;
  41.     }
  42.     public int getNum() {
  43.         return num;
  44.     }
  45.     public void setNum(int num) {
  46.         this.num = num;
  47.     }
  48.     public double getWeight() {
  49.         return weight;
  50.     }
  51.     public void setWeight(double weight) {
  52.         this.weight = weight;
  53.     }
  54. }
server包，
Server类，启动监听，
  1. package mina.server;
  3. import java.net.InetSocketAddress;
  5. import org.apache.mina.common.DefaultIoFilterChainBuilder;
  6. import org.apache.mina.common.IoAcceptor;
  7. import org.apache.mina.common.IoAcceptorConfig;
  8. import org.apache.mina.transport.socket.nio.SocketAcceptor;
  9. import org.apache.mina.transport.socket.nio.SocketAcceptorConfig;
  10. import org.apache.mina.transport.socket.nio.SocketSessionConfig;
  12. public class Server {
  14.     /** Choose your favorite port number. */
  15.     private static final int PORT = 8080;
  17.     private static final boolean USE_SSL = false;
  19.     public static void main(String[] args) throws Exception {
  20.         IoAcceptor acceptor = new SocketAcceptor();
  21.         IoAcceptorConfig config = new SocketAcceptorConfig();
  22.         DefaultIoFilterChainBuilder chain = config.getFilterChain();
  24.         ((SocketSessionConfig) config.getSessionConfig()).setReuseAddress(true);
  25.         // Add SSL filter if SSL is enabled.
  26.         if (USE_SSL) {
  27.             // addSSLSupport(chain);
  28.         }
  30.         // Bind
  31.         acceptor.bind(new InetSocketAddress(PORT), new ServerHandler(), config);
  33.         System.out.println("Listening on port " + PORT);
  34.     }
  35. }
RpcServerImpl实现类，实现common包中定义的接口，实际的处理的地方。
  1. package mina.server;
  3. import mina.common.Apple;
  4. import mina.common.RpcInterface;
  6. public class RpcServerImpl implements RpcInterface {
  8.     public String getStringValue(String arg0,int arg1,Apple arg2) {
  9.         System.out.println("apple time is "+arg2.getPdate());
  10.         return "this is sign from server.";
  11.     }
  13.     public int getIntValue() {
  14.         return 3;
  15.     }
  17.     public void printPrice() {
  18.         System.out.println("******price**********");
  19.         System.out.println("$10000000000000000000");
  20.         System.out.println("*********************");
  21.     }
  22. }
ServerHandler类，处理客户端连接事务，和客户端通信处理的类。
  1. package mina.server;
  3. import java.io.InputStream;
  4. import java.io.OutputStream;
  5. import java.lang.reflect.InvocationTargetException;
  6. import java.lang.reflect.Method;
  7. import java.net.SocketAddress;
  8. import java.util.ArrayList;
  9. import java.util.Enumeration;
  10. import java.util.Hashtable;
  11. import java.util.Iterator;
  12. import java.util.List;
  13. import java.util.Properties;
  15. import org.apache.mina.common.ByteBuffer;
  16. import org.apache.mina.common.IoSession;
  17. import org.apache.mina.handler.StreamIoHandler;
  18. import org.apache.mina.integration.jmx.IoSessionManager;
  20. public class ServerHandler extends StreamIoHandler {
  22.     Hashtable sessionMgr = new Hashtable();
  23.     List serverList = new ArrayList();
  24.     ServerHandler() {
  25.         serverList.add(new RpcServerImpl());
  26.     }
  28.     public void messageReceived(IoSession session, Object buf) {
  29.         SocketAddress adr = session.getRemoteAddress();
  30.         System.out.println("remote address is =" + adr.toString());
  31.         System.out.println("buf=" + buf.toString());
  32.         if (buf instanceof ByteBuffer) {
  33.             ByteBuffer bb = (ByteBuffer) buf;
  34.             System.out.println("bbb===" + bb);
  35.             try {
  36.                 Properties prop = (Properties) bb.getObject();
  37.                 System.out.println("prop==" + prop);
  39.                 String interfaceName = (String) prop.get("interface");
  40.                 Iterator it = serverList.iterator();
  41.                 while (it.hasNext()) {// 查找实例
  42.                     Object serobj = it.next();
  43.                     Class[] clazz = serobj.getClass().getInterfaces();
  44.                     if (isContains(clazz, interfaceName)) {// 找到相应实例
  45.                         System.out.println("find.." + interfaceName);
  46.                         int argc = Integer.parseInt(String.valueOf(prop
  47.                                 .get("argc")));
  48.                         Class[] types = null;// 参数类型数组
  49.                         Object[] args = null;// 参数对象数组
  51. //                       不为最大值时表示有正常参数，否则为无参数。
  52.                         if (argc != Integer.MAX_VALUE) {
  53.                             // 重组参数列表
  54.                             types = new Class[argc];
  55.                             args = new Object[argc];
  56.                             List typeList = (List) prop.get("types");
  57.                             List argList = (List) prop.get("args");
  58.                             for (int i = 0; i < argc; i++) {
  59.                                 args[i] = argList.get(i);
  60.                                 types[i] = (Class) typeList.get(i);
  61.                                 System.out.println("arg###=" + args[i]
  62.                                         + "   type@@@=" + types[i]);
  63.                             }
  64.                         }
  65.                         String methodName = prop.getProperty("method");
  67.                         Method method = serobj.getClass().getMethod(methodName,
  68.                                 types);
  69.                         //调用
  70.                         Object resultObject = method.invoke(serobj, args);
  72.                         //回复
  73.                         replyCall(session.getRemoteAddress().toString(),resultObject);
  74.                     }
  75.                 }
  76.             } catch (Exception e) {
  77.                 e.printStackTrace();
  78.             }
  79.         }
  80.     }
  81.     
  82.     //广播操作，广播消息或事件.标记为"BROADCAST" 
  83.     public void brocastMessage(Object obj){
  84.         Enumeration enu = sessionMgr.keys();
  85.         while (enu.hasMoreElements()) {
  86.             String addr = (String) enu.nextElement();
  87.             sendMessage("BROADCAST",addr,obj);
  88.         }
  89.     }
  90.     
  91.     //通知操作，通知消息或事件。标记为"NOTIFY"
  92.     public void notify(String addr,Object obj){
  93.         sendMessage("NOTIFY",addr,obj);
  94.     }
  95.     
  96.     //响应RPC调用。
  97.     public void replyCall(String addr,Object obj){
  98.         sendMessage("REPLYCALL",addr,obj);
  99.     }
  100.     
  101.     private void sendMessage(String mark,String addr,Object obj){
  102.         Object sessionObj = sessionMgr.get(addr);
  103.         if(sessionObj!=null){
  104.             IoSession session = (IoSession)sessionObj;
  105.             if(!session.isConnected()){
  106.                 return;
  107.             }
  108.             Properties resultProp = new Properties();
  109.             resultProp.setProperty("mark",mark);
  110.             if(obj!=null) //对于无返回的情况。
  111.                 resultProp.put("Object",obj);
  112.             ByteBuffer bb = ByteBuffer.allocate(16);
  113.             bb.setAutoExpand(true);
  114.             bb.putObject(resultProp);
  115.             bb.flip();
  116.             session.write(bb);
  117.         }else{
  118.             System.out.println("session null.addr="+addr);
  119.         }
  120.     }
  121.     
  122.     protected void processStreamIo(IoSession session, InputStream ins,
  123.             OutputStream ous) {
  124.         System.out.println("processStreamIo is called.");
  125.     }
  127.     private boolean isContains(Class[] clazz, String ifName) {
  128.         for (int i = 0; i < clazz.length; i++) {
  129.             if (clazz[i].getName().equals(ifName))
  130.                 return true;
  131.         }
  132.         return false;
  133.     }
  135.     public void sessionOpened(IoSession ssn) {
  136.         System.out.println("session open for " + ssn.getRemoteAddress());
  137.         sessionMgr.put(ssn.getRemoteAddress().toString(),ssn);
  138.     }
  140.     public void exceptionCaught(IoSession ssn, Throwable cause) {
  141.         cause.printStackTrace();
  142.         sessionMgr.remove(ssn.getRemoteAddress().toString());
  143.         ssn.close();
  144.     }
  146.     public void sessionClosed(IoSession ssn) throws Exception {
  147.         System.out.println("session closed from " + ssn.getRemoteAddress());
  148.         sessionMgr.remove(ssn.getRemoteAddress().toString());
  149.     }
  151. }
client包，实现client功能 。连接服务端调用common接口提供的方法。
Client类，连接服务端，调用客户端的方法。
  1. package mina.client;
  3. import java.net.InetSocketAddress;
  5. import mina.common.Apple;
  7. import org.apache.mina.common.ConnectFuture;
  8. import org.apache.mina.common.IoConnectorConfig;
  9. import org.apache.mina.transport.socket.nio.SocketConnector;
  11. public class Client {
  12.     public static void main(String[] args) throws Exception {
  14.         // Create TCP/IP connector.
  15.         SocketConnector connector = new SocketConnector();
  17.         // Set connect timeout.
  18.         ((IoConnectorConfig) connector.getDefaultConfig())
  19.                 .setConnectTimeout(15);
  21.         ClientIoHandler handler = new ClientIoHandler();
  23.         // Start communication.
  24.         ConnectFuture cf = connector.connect(new InetSocketAddress("localhost",
  25.                 8080), handler);
  27.         // Wait for the connection attempt to be finished.
  28.         System.out.println("start to join");
  29.         cf.join();
  30.         cf.getSession();
  32.         System.out.println("test get value="
  33.                 + RpcClientImpl.getInstance().getStringValue("adsf", 222,
  34.                         new Apple()));
  36.         System.out.println("test get int value="
  37.                 + RpcClientImpl.getInstance().getIntValue());
  39.         System.out.println("test call process start.");
  40.         RpcClientImpl.getInstance().printPrice();
  41.         System.out.println("test call process end.");
  42.     }
  43. }

AbstractClientImpl类，为client实现接口时使用的超类。
  1. package mina.client;
  3. import org.apache.mina.common.IoSession;
  5. public abstract class AbstractClientImpl {
  6.     
  7.     IoSession session = null;
  8.     ClientIoHandler handler = null;
  9.     public IoSession getSession() {
  10.         return session;
  11.     }
  13.     public void setSession(IoSession session) {
  14.         this.session = session;
  15.     }
  17.     public ClientIoHandler getHandler() {
  18.         return handler;
  19.     }
  21.     public void setHandler(ClientIoHandler handler) {
  22.         this.handler = handler;
  23.     }
  24. }
RpcClientImpl类，客户端实现common中接口的类，实际上是一种伪实现，在各个方法中组装调用的接品，方法，参数等信息，传到服务端由服务端的通信处理类处理信息，实现服务端调用并返回结果。
  1. package mina.client;
  3. import java.util.ArrayList;
  4. import java.util.Date;
  5. import java.util.List;
  6. import java.util.Properties;
  8. import mina.common.Apple;
  9. import mina.common.RpcInterface;
  11. import org.apache.mina.common.ByteBuffer;
  13. public class RpcClientImpl extends AbstractClientImpl implements RpcInterface {
  15.     private static String IFNAME = RpcInterface.class.getName();
  16.     
  17.     private static RpcClientImpl instance = null;
  18.     
  19.     public static RpcClientImpl getInstance(){
  20.         if(instance == null){
  21.             instance = new RpcClientImpl();
  22.         }
  23.         return instance;
  24.     }
  25.     
  26.     public String getStringValue(String arg0,int arg1,Apple arg2) {
  27.         if(session!=null){
  28.             Properties prop = new Properties();
  29.             prop.setProperty("interface",IFNAME);
  30.             prop.setProperty("method","getStringValue");
  31.             prop.put("argc","3");//参数个数.
  32.         
  33.             //生成参数类型链表
  34.             List typeList = new ArrayList();
  35.             typeList.add(String.class);
  36.             typeList.add(Integer.TYPE);
  37.             typeList.add(Apple.class);
  38.             
  39.             //生成参数对象链表
  40.             List argList = new ArrayList();
  41.             argList.add(arg0);
  42.             argList.add(arg1);
  43.             argList.add(arg2);
  44.             
  45.             prop.put("types",typeList);
  46.             prop.put("args",argList);
  47.             
  48.             ByteBuffer bb = ByteBuffer.allocate(16);
  49.             bb.setAutoExpand( true );
  50.             bb.putObject(prop);
  51.             bb.flip();
  52.             System.out.println("bbb==="+bb.toString());
  53.             session.write(bb);
  54.         }
  55.         
  56.         try {
  57.             System.out.println("handler="+handler);
  58.             System.out.println("lock="+handler.lock);
  59.             printStamp("1");
  60.             synchronized(handler.lock){
  61.                 printStamp("2");
  62.                 handler.lock.wait();
  63.                 printStamp("3");
  64.                 if(handler.resultObject!=null){
  65.                     return String.valueOf(handler.resultObject);
  66.                 }
  67.             }
  68.         } catch (InterruptedException e) {
  69.             e.printStackTrace();
  70.         }
  71.         
  72.         return null;
  73.     }
  75.     public int getIntValue() {
  76.         if(session!=null){
  77.             Properties prop = new Properties();
  78.             prop.setProperty("interface",IFNAME);
  79.             prop.setProperty("method","getIntValue");
  80.             prop.put("argc",Integer.MAX_VALUE);
  81.             //prop.put("object",new Object());
  82.             ByteBuffer bb = ByteBuffer.allocate(16);
  83.             bb.setAutoExpand( true );
  84.             bb.putObject(prop);
  85.             bb.flip();
  86.             System.out.println("bbb==="+bb.toString());
  87.             session.write(bb);
  88.         }
  89.         try {
  90.             System.out.println("handler="+handler);
  91.             System.out.println("lock="+handler.lock);
  92.             printStamp("1");
  93.             synchronized(handler.lock){
  94.                 printStamp("2");
  95.                 handler.lock.wait();
  96.                 printStamp("3");
  97.                 if(handler.resultObject!=null){
  98.                     return Integer.parseInt(String.valueOf(handler.resultObject));
  99.                 }
  100.             }
  101.         } catch (InterruptedException e) {
  102.             e.printStackTrace();
  103.         }
  104.         
  105.         return Integer.MIN_VALUE;
  106.     }
  108.     
  109.     public void printStamp(String str){
  110.         System.out.println((new Date().toString())+" "+str);
  111.     }
  113.     public void printPrice() {
  114.         if(session!=null){
  115.             Properties prop = new Properties();
  116.             prop.setProperty("interface",IFNAME);
  117.             prop.setProperty("method","printPrice");
  118.             prop.put("argc",Integer.MAX_VALUE);//参数个数
  119.             
  120.             ByteBuffer bb = ByteBuffer.allocate(16);
  121.             bb.setAutoExpand( true );
  122.             bb.putObject(prop);
  123.             bb.flip();
  124.             System.out.println("bbb==="+bb.toString());
  125.             session.write(bb);
  126.         }
  127.         
  128.         try {
  129.             System.out.println("handler="+handler);
  130.             System.out.println("lock="+handler.lock);
  131.             printStamp("1");
  132.             synchronized(handler.lock){
  133.                 printStamp("2");
  134.                 handler.lock.wait();
  135.                 printStamp("3");
  136.                 return;
  137.             }
  138.         } catch (InterruptedException e) {
  139.             e.printStackTrace();
  140.         }
  141.         return;
  142.     }
  143. }
ClientIoHandler类，处理服务端通信的类，用于和服务端建立通信，处理由服务端发送过来的消息。
  1. package mina.client;
  3. import java.io.InputStream;
  4. import java.io.OutputStream;
  5. import java.util.ArrayList;
  6. import java.util.Iterator;
  7. import java.util.List;
  8. import java.util.Properties;
  10. import org.apache.mina.common.ByteBuffer;
  11. import org.apache.mina.common.IdleStatus;
  12. import org.apache.mina.common.IoSession;
  13. import org.apache.mina.handler.StreamIoHandler;
  15. public class ClientIoHandler extends StreamIoHandler{
  17.     IoSession session = null;
  18.     
  19.     List implList = new ArrayList();
  20.     
  21.     Object resultObject;
  22.     Object lock = new Object();
  23.     
  24.     public ClientIoHandler(){
  25.         //init impls.
  26.         RpcClientImpl impl = RpcClientImpl.getInstance();
  27.         implList.add(impl);
  28.         //impl.setHandler(this);
  29.     }
  30.     
  31.     private void initImpls(){
  32.         Iterator it = implList.iterator();
  33.         while(it.hasNext())
  34.         {
  35.             Object obj = it.next();
  36.             if(obj instanceof AbstractClientImpl){
  37.                 AbstractClientImpl impl = (AbstractClientImpl)obj;
  38.                 impl.setSession(session);
  39.                 System.out.println("add this handler");
  40.                 impl.setHandler(this);
  41.             }
  42.         }
  43.     }
  44.     
  45.     //收到服务端消息后的处理，框架内部为异步。在应用中改为同步。
  46.     public void messageReceived(IoSession session, Object buf) {
  47.         System.out.println("receive message.");
  48.         
  49.         System.out.println(buf.toString());
  50.         try{
  51.             if(buf instanceof ByteBuffer){
  52.                 Properties prop = (Properties)(((ByteBuffer)buf).getObject());
  53.                 System.out.println("received prop="+prop);
  54.                 String mark = prop.getProperty("mark");
  55.                 //根据mark类型，选择不同处理。
  56.                 if(mark.equals("REPLYCALL")){
  57.                     Object obj = prop.get("Object");
  58.                     synchronized(lock){
  59.                         System.out.println("result="+obj);
  60.                         resultObject = obj;
  61.                         lock.notify();
  62.                     }
  63.                 }else if(mark.equals("BROADCAST")){
  64.                     Object obj = prop.get("Object");
  65.                     System.out.println("BROADCAST obj="+obj);
  66.                 }else if(mark.equals("NOTIFY")){
  67.                     Object obj = prop.get("Object");
  68.                     System.out.println("NOTIFY obj="+obj);
  69.                 }else{
  70.                     System.out.println("unknow mark.mark="+mark);
  71.                 }
  72.             }else{
  73.                 System.out.println("class type error.");
  74.             }
  75.         } catch (ClassNotFoundException e) {
  76.             e.printStackTrace();
  77.         }
  78.     }
  79.     
  80.     protected void processStreamIo(IoSession session, InputStream is, OutputStream os) {
  81.         System.out.println("process stream info,"+session.getRemoteAddress());
  82.     }
  84.     public void sessionOpened(IoSession session) {
  85.         // Set reader idle time to 10 seconds.
  86.         // sessionIdle(...) method will be invoked when no data is read
  87.         // for 10 seconds.
  88.        System.out.println("open session..");
  89.         this.session = session;
  90.        initImpls();
  91.        session.setIdleTime(IdleStatus.READER_IDLE, 10);
  92.     }
  94.     public void sessionClosed(IoSession session) {
  95.         // Print out total number of bytes read from the remote peer.
  96.         System.err.println("Total " + session.getReadBytes() + " byte(s)");
  97.     }
  99.     public void sessionIdle(IoSession session, IdleStatus status) {
  100.         // Close the connection if reader is idle.
  101.         if (status == IdleStatus.READER_IDLE)
  102.             session.close();
  103.     }
  104. }

需要改进的地方：
    在实际应用中，还需要精简客户端的结构，实现多个impl，一个handler，即实现一个将消息dispatch方法。服务端对客户端的主动消息还需要另外定义格式，以满足不同条件下的需要。或者需要定义内容比较详细的事件。





最后

以上就是动人战斗机为你收集整理的使用Mina实现RPC调用，消息通知，广播。的全部内容，希望文章能够帮你解决使用Mina实现RPC调用，消息通知，广播。所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(80)

相关文章

Dubbo学习笔记(三) RPC核心原理和线程模型
Dubbo学习笔记(三) RPC核心原理和线程模型
RPC框架实现原理
RPC框架实现原理
RPC 工作原理一、RPC二、RPC的原理和框架三、Java中常用的RPC框架四、RPC和消息队列的差异五、RPC框架的核心技术点
RPC 工作原理一、RPC二、RPC的原理和框架三、Java中常用的RPC框架四、RPC和消息队列的差异五、RPC框架的核心技术点
RPC的实现原理
RPC的实现原理
使用Mina实现RPC调用，消息通知，广播。
使用Mina实现RPC调用，消息通知，广播。
RPC框架实现原理，看这篇就够了！
RPC框架实现原理，看这篇就够了！
Liunx 服务器上编译安装swool扩展和Swoole简单实现RPC原理
Liunx 服务器上编译安装swool扩展和Swoole简单实现RPC原理
rpc框架实现原理
rpc框架实现原理

评论列表共有 0 条评论

立即
投稿
返回
顶部