Netty RPC demo 试跑

阅读量：4081 次

发布时间：2019-05-25

本文共 12514 字，大约阅读时间需要 41 分钟。

（一）

http://www.cnblogs.com/jietang/p/5615681.html

提供了一个Netty＋Protobuf的RPC解决方案，并提供了demo：

clone该demo，maven编译，有一个ojdbc6无法下载，查找资料

由于需要oracle官方授权，所以maven上无法下载ojdbc，需要自己下载，然后通过命令加载到本地maven库中，详细步骤如下

1、到官方下载，地址：，找到“drivers”-“jdbc Drivers”，打开，点击同意协议，就可以选择版本下载了

2、假设下载的是11.2.0.1.0，放在本地

3、打开命令行窗口，执行以下命令加载到本地

[plain]
 mvn install:install-file -DgroupId=com.oracle -DartifactId=ojdbc6 -Dversion=11.2.0.1.0 －Dpackaging=jar -Dfile=ojdbc6.jar  

重新编译 ok ，target目录下生成

NettyRPC-1.0-SNAPSHOT.jar

运行该服务

java -jar <jar-file-name>.jar

_ _

| | | |

_ __ ___| |_| |_ _ _ _ __ _ __ ___

| '_ \ / _ \ __| __| | | | '__| '_ \ / __|

| | | | __/ |_| |_| |_| | | | |_) | (__

|_| |_|\___|\__|\__|\__, |_| | .__/ \___|

__/ | | |

|___/ |_|

[NettyRPC 2.0,Build 2016/10/7,Author:tangjie http://www.cnblogs.com/jietang/]

[author tangjie] Netty RPC Server start success!

ip:127.0.0.1

port:18887

protocol:You can open your web browser see NettyRPC server api interface: http://127.0.0.1:18886/NettyRPC.html

RpcSerializeProtocol[serializeProtocol=protostuff,name=PROTOSTUFFSERIALIZE,ordinal=3]

然后运行客户端：

com.newlandframework.test.RpcParallelTest

客户端显示：

......

calc multi result:[992016]乘法计算RPC调用总共耗时: [974] 毫秒[author tangjie] Netty RPC Server 消息协议序列化第[0]轮并发验证结束!

服务端显示：

....... RPC Server Send message-id respone:6abe54e6-d9ef-44d7-ab4e-28a86f817dd3 RPC Server Send message-id respone:6d3a3e1c-198a-4c95-9df4-6af32f571dec RPC Server Send message-id respone:aa46cf22-7a0c-4c47-afc1-0a68e682bd75 RPC Server Send message-id respone:7f04cc95-4c3a-483f-9322-005ae9c3082e RPC Server Send message-id respone:cdca0293-404b-4dbf-a11d-2be9a15ca7ee RPC Server Send message-id respone:bc9144b8-3d50-438d-92df-8ef8ec6d4255

done

（二）10.16 调试其jdbc部分（NettyRpcJdbcServerTest）

首先需要数据库支持，更换掉oracle的ojdbc6，换为mysql引擎：


               
    
     mysql
                
    
     mysql-connector-java
                
    
     5.1.6

我们首先要编译运行/test/jdbc/NettyRpcJdbcServerTest.java:

public class NettyRpcJdbcServerTest {    // FIXME: 2017/9/25 确保先启动NettyRPC服务端应用:NettyRpcJdbcServerTest，再运行NettyRpcJdbcClientTest、NettyRpcJdbcClientErrorTest！    public static void main(String[] args) {        new ClassPathXmlApplicationContext("classpath:rpc-invoke-config-jdbc-server.xml");    }}

找到rpc-invoke-config-jdbc-server.xml: 更换其中的datasource bean：

原：

改为mysql的：

在mysql中建立Person表

CREATE TABLE `person` (  `id` int(11) NOT NULL AUTO_INCREMENT,  `name` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '',  `age` int(11) NOT NULL,  `birthday` datetime DEFAULT NULL,  PRIMARY KEY (`id`)) ENGINE=MyISAM AUTO_INCREMENT=2 DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='test';

编译运行

Server，运行NettyRpcJdbcClientTest

客户端显示：

call pojo rpc result:0

---------------------------------------------

Person <<id:1 name:小好 age:2 birthday:Tue Aug 11 16:47:00 CST 2015>>

Person <<id:2 name:小好 age:2 birthday:Tue Aug 11 16:47:00 CST 2015>>

Process finished with exit code 0

done，

期间，要更换 oracle的to_date 为mysql的str_to_date函数

（三）10.20

加入log4j

参考：http://harborchung.iteye.com/blog/2271509

在主函数中添加

PropertyConfigurator.configure("classes/log4j.properties");

（四）11.13 加入自动装配jdbctemplate，aop统一处理异常处理，事务支持

首先来看自动装配jdbctemplate，此前是这样的：

JdbcTemplate template = new JdbcTemplate(this.dataSource);

public class MyPojoImpl implements MyPojo {    private DataSource dataSource;    public void setDataSource(DataSource dataSource) {        this.dataSource = dataSource;    }

利用setter注入MyPojoImpl的依赖项dataSource，然后直接实例化JdbcTemplae，这样就使jdbctemplate在spring托管之外

现改为：

public class MyPojoImpl implements MyPojo {//    private DataSource dataSource;    @Autowired    private JdbcTemplate jdbcTemplate;    private static Logger LOG = Logger.getLogger(MyPojoImpl.class);//    public void setDataSource(DataSource dataSource) {//        this.dataSource = dataSource;//    }

运行，报错：jdbcTemplate为null，装配失败

参考：http://www.iteye.com/problems/92069

“

<context：annotation-config />
这句话才是激活Bean的属性上的自动注入相关的注释处理：Autowired之类

<context：component-scan base-package =“sunships.dhcc”/>
这个只是搜索类，并按照类声明的注解作为合适的bean加入ApplicationContext

”

在spring环境中加入：

运行ok，这个注解是告诉spring加载预定义的一些bean，Autowired在其中

然后处理aop统一异常处理：

添加：

@Aspectpublic class AuthInterceptor {    private static Logger LOG = Logger.getLogger(AuthInterceptor.class);    @Pointcut("execution(* com.newlandframework.rpc.services.impl.*.*(..))")    private void aroundMethod(){}//定义一个切入点    @Around("aroundMethod()")    public Object doBasicProfiling(ProceedingJoinPoint pjp) throws Throwable {        Object object = null;        try {            System.out.println("aop start");            object = pjp.proceed();//执行该方法        } catch (Exception e) {            System.out.println("aop exception");            e.printStackTrace();            LOG.error("global error:", e);        }        System.out.println("aop end");        //    System.out.println("退出方法");        return object;    }}

注入：

idea中aop标签报错，参考：http://blog.csdn.net/qq_37062668/article/details/74298491

添加

xmlns:aop="http://www.springframework.org/schema/aop"

mvn编译通过，运行，报错：

通配符的匹配很全面, 但无法找到元素 'aop:aspectj-autoproxy' 的声明。

在

xsi:schemaLocation中添加：

http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.3.xsd

编译运行：

aop start

jdbc Tao query!

[INFO] {RpcThreadPool-thread-1} com.newlandframework.rpc.services.impl.MyPojoImpl.queryList(66) | MyPojo.queryList done.

aop end

RPC Server Send message-id respone:4964cc58-3fd8-4385-83b8-3ee552165259

通过。

期间在pointcut上的报错

错误: warning no match for this type name: com.zrm.service [Xlint:invalidAbsoluteTypeName]

处理参考了http://blog.csdn.net/yangxiaovip/article/details/31797475

由

@Pointcut("execution(* com.newlandframework.rpc.services.impl.*(..))")

改为：

@Pointcut("execution(* com.newlandframework.rpc.services.impl.*.*(..))")

最后，调试异常事务回滚

@Aspectpublic class AuthInterceptor {    private static Logger LOG = Logger.getLogger(AuthInterceptor.class);    @Pointcut("execution(* com.newlandframework.rpc.services.impl.*.*(..))")    private void aroundMethod(){}//定义一个切入点    @Around("aroundMethod()")    public Object doBasicProfiling(ProceedingJoinPoint pjp) throws Throwable {        Object object = null;        try {            System.out.println("aop start");            object = pjp.proceed();//执行该方法        } catch (Exception e) {            System.out.println("aop exception");            e.printStackTrace();            LOG.error("global error:", e);            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();        }        System.out.println("aop end");        //    System.out.println("退出方法");        return object;    }}

主要是加入

TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();

事务处理器注入

特意建立函数：

@Transactional    @Override    public List
   
     queryListError() {        System.out.println("jdbc Tao query!");        String insertSql = "insert into tao (col2) values ('call exception')";        jdbcTemplate.execute(insertSql);        String sql = "select * from tao2";     //   JdbcTemplate template = new JdbcTemplate(this.dataSource);                List
    
     
      > rows = jdbcTemplate.queryForList(sql);        return null;    }

先插入，后执行错误sql语句，加上Transactional注解

aop startjdbc Tao query!aop exceptionorg.springframework.jdbc.BadSqlGrammarException: StatementCallback; bad SQL grammar [select * from tao2]; nested exception is com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Table 'amac_data_test.tao2' doesn't exist	at org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator.doTranslate(SQLErrorCodeSQLExceptionTranslator.java:231)	at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:73)	at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:413)	at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:468)	at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:478)	at org.springframework.jdbc.core.JdbcTemplate.queryForList(JdbcTemplate.java:518)	at com.newlandframework.rpc.services.impl.MyPojoImpl.queryListError(MyPojoImpl.java:83)

且回滚成功，并未插入数据

此过程中

@Transactional

注解和,

TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();

缺一不可，done

（五）12.21 netty客户端与web项目的结合

1.最初，每个请求都遵循

@RequestMapping(value = "testNetty", method = RequestMethod.POST)    @ResponseBody    @ApiImplicitParams({})    @ApiOperation(value="testNetty")    public Object testNetty() {        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:netty.xml");        MyPojo manage = (MyPojo) context.getBean("MyPojoJdbc");        List
   
     list = null;        try {            list = manage.queryList();            for (int i = 0; i < list.size(); i++) {                System.out.println(list.get(i));            }        } catch (Exception e) {            e.printStackTrace();        } finally {            context.destroy();        }        return list;    }

每次请求完都释放netty的context，

context.destroy()导致

ClientStopEventListener

被激发，如下：

public class ClientStopEventListener {    public int lastMessage = 0;    @Subscribe    public void listen(ClientStopEvent event) {        lastMessage = event.getMessage();        // sunyuming        MessageSendExecutor.getInstance().stop();    }    public int getLastMessage() {        return lastMessage;    }}     public void stop() {        loader.unLoad();    }     public void unLoad() {        messageSendHandler.close();        threadPoolExecutor.shutdown();        eventLoopGroup.shutdownGracefully();    } 其中， private static ListeningExecutorService threadPoolExecutor = MoreExecutors.listeningDecorator((ThreadPoolExecutor) RpcThreadPool.getExecutor(threadNums, queueNums)); 静态线程池反复操作释放后挂了

故有了2

2.

public class ClientStopEventListener {    public int lastMessage = 0;    @Subscribe    public void listen(ClientStopEvent event) {        lastMessage = event.getMessage();        // sunyuming     //   MessageSendExecutor.getInstance().stop();    }    public int getLastMessage() {        return lastMessage;    }} @WebListenerpublic class MyServletContextListener implements ServletContextListener {    @Override    public void contextDestroyed(ServletContextEvent arg0) {        MessageSendExecutor.getInstance().stop();    }}

监听spring web context的销毁，在其中激发各种线程资源的回收，取消对netty context销毁的监控中stop函数，这样，线程资源只会回收一次，不会报错，也跑起来了

但每次请求都创建netty context太耗费资源，显然不行，于是有了3

3.

public static final ClassPathXmlApplicationContext NETTY_CONTEXT = new ClassPathXmlApplicationContext("classpath:netty.xml"); 这样就只创建一次，销毁时 @WebListenerpublic class MyServletContextListener implements ServletContextListener {    @Override    public void contextDestroyed(ServletContextEvent arg0) {        System.out.println("netty销毁");        // sunyuming        OrgNewController.NETTY_CONTEXT.destroy();        MessageSendExecutor.getInstance().stop();    }}

ok，正常允许，但其实已经埋下了隐患：

正常的销毁顺序应该为

spring context destroy 中触发 netty context destroy 再触发 MassageSendExecutor.getInstance().stop()

此时的代码为：

spring context destory 中直接同时触发MassageSendExecutor.getInstance().stop()，再触发 netty context destroy ，其中MassageSendExecutor.getInstance().stop()由原本在netty context destroy中，移动到 spring web context destroy中直接处理了

然后就发生隐患了

4.有2个web项目要用到netty client，在操作第二个时，代码复制过去后，出现tomcat shutdown之后没有关闭进程，根据以往的经验：http://blog.csdn.net/silyvin/article/details/72678375

直接吧问题定位到线程池未shutdown导致内存泄露

回想此时的代码：

web2这个项目没有监听spring context destroy

复制过去的netty代码中缺少MassageSendExecutor.getInstance().stop()，（因为被移动到web1的spring context destroy中去了），所以线程没有释放，导致泄露

解决方案：

5.于是回归规范，将自行在netty client 中监听netty context销毁的代码中恢复MassageSendExecutor.getInstance().stop()

public class ClientStopEventListener {    public int lastMessage = 0;    @Subscribe    public void listen(ClientStopEvent event) {        lastMessage = event.getMessage();        // sunyuming        MessageSendExecutor.getInstance().stop();    }    public int getLastMessage() {        return lastMessage;    }}

在web1中，去除对MassageSendExecutor.getInstance().stop()的显式直接操作：

@Override    public void contextDestroyed(ServletContextEvent arg0) {        System.out.println("netty销毁");        // sunyuming        OrgNewController.NETTY_CONTEXT.destroy();    //    MessageSendExecutor.getInstance().stop();    }

这样就经由正确的路线去释放资源 spring web context destroy －－》 netty context destroy －－》MassageSendExecutor.getInstance().stop()

在web2中也添加对spring web context destroy 的监听

解决

你可能感兴趣的文章