Netty 编解码技术 数据通信和心跳监控案例
多台服务器之间在进行跨进程服务调用时，需要使用特定的编解码技术，对需要进行网络传输的对象做编码和解码操作，以便完成远程调用。Netty提供了完善，易扩展，易使用的编解码技术。本章除了介绍Marshalling的使用，还会基于编解码技术实现数据通信和心跳检测案例。通过本章，你将学到Java序列化的优缺点，主流编解码框架的特点，模拟特殊长连接通信，心跳监控案例。还在等什么，丰满的知识等你来拿！
技术：编解码，数据通信，心跳监控
说明：github上有完整代码，部分文字描述摘录《Netty权威指南》 源码：https://github.com/ITDragonBlog/daydayup/tree/master/Netty/netty-stu编解码
Netty 的一大亮点就是使用简单，将常用的功能和API进行了很好的封装，编解码也不例外。针对编解码功能，Netty提供了通用的编解码框架和常用的编解码类库，方便用户扩展和使用。从而降低用户的工作量和开发门槛。在io.netty.handler.codec目录下找到很多预置的编解码功能.
其实在，就已经使用了Netty的编解码技术，如：DelimiterBasedFrameDecoder，FixedLengthFrameDecoder，StringDecoder什么是编解码技术
编码（Encode）也称序列化（serialization），将对象序列化为字节数组，用于网络传输、数据持久化等用途。
解码（Decode）也称反序列化（deserialization），把从网络、磁盘等读取的字节数组还原成原始对象，以方便后续的业务逻辑操作。主流编解码框架
Java序列化
Java序列化使用简单，开发难度低。只需要实现java.io.Serializable接口并生成序列化ID，这个类就能够通过java.io.ObjectInput序列化和java.io.ObjectOutput反序列化。
但它也有存在很多缺点 : 1 无法跨语言（java的序列化是java语言内部的私有协议，其他语言并不支持）， 2 序列化后码流太大（采用二进制编解码技术要比java原生的序列化技术强）， 3 序列化性能太低JBoss的Marshalling
JBoss的Marshalling是一个Java对象的序列化API包，修正了JDK自带序列化包的很多问题，又兼容java.io.Serializable接口；同时可通过工厂类进行参数和特性地配置。
1) 可插拔的类解析器，提供更加便捷的类加载定制策略，通过一个接口即可实现定制； 2) 可插拔的对象替换技术，不需要通过继承的方式； 3) 可插拔的预定义类缓存表，可以减小序列化的字节数组长度，提升常用类型的对象序列化性能； 4) 无须实现java.io.Serializable接口，即可实现Java序列化； 5) 通过缓存技术提升对象的序列化性能。 6) 使用范围小，通用性较差。Google的Protocol Buffers
Protocol Buffers由谷歌开源而来。将数据结构以 .proto 文件进行描述，通过代码生成工具可以生成对应数据结构的POJO对象和Protobuf相关的方法和属性。
1) 结构化数据存储格式（XML，JSON等）； 2) 高效的编解码性能； 3) 平台无关、扩展性好； 4) 官方支持Java、C++和Python三种语言。MessagePack框架
MessagePack是一个高效的二进制序列化格式。和JSON一样跨语言交换数据。但是它比JSON更快、更小（It's like JSON.but fast and small）。
1) 高效的编解码性能； 2) 跨语言； 3) 序列化后码流小；Marshalling 配置工厂
package com.itdragon.marshalling;import io.netty.handler.codec.marshalling.DefaultMarshallerProvider;import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider;import io.netty.handler.codec.marshalling.MarshallerProvider;import io.netty.handler.codec.marshalling.MarshallingDecoder;import io.netty.handler.codec.marshalling.MarshallingEncoder;import io.netty.handler.codec.marshalling.UnmarshallerProvider;import org.jboss.marshalling.MarshallerFactory; import org.jboss.marshalling.Marshalling; import org.jboss.marshalling.MarshallingConfiguration; public final class ITDragonMarshallerFactory { private static final String NAME = "serial"; // serial表示创建的是 Java序列化工厂对象.由jboss-marshalling-serial提供 private static final Integer VERSION = 5; private static final Integer MAX_OBJECT_SIZE = 1024 * 1024 * 1; // 单个对象最大长度 /** * 创建Jboss Marshalling 解码器MarshallingDecoder */ public static MarshallingDecoder buildMarshallingDecoder() { // step1 通过工具类 Marshalling，获取Marshalling实例对象，参数serial 标识创建的是java序列化工厂对象 final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory(NAME); // step2 初始化Marshalling配置 final MarshallingConfiguration configuration = new MarshallingConfiguration(); // step3 设置Marshalling版本号 configuration.setVersion(VERSION); // step4 初始化生产者 UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration); // step5 通过生产者和单个消息序列化后最大长度构建 Netty的MarshallingDecoder MarshallingDecoder decoder = new MarshallingDecoder(provider, MAX_OBJECT_SIZE); return decoder; } /** * 创建Jboss Marshalling 编码器MarshallingEncoder */ public static MarshallingEncoder builMarshallingEncoder() { final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory(NAME); final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(VERSION); MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration); MarshallingEncoder encoder = new MarshallingEncoder(provider); return encoder; } } 数据通信
一个网络应用最重要的工作莫过于数据的传输。两台机器之间如何建立连接才能提高服务器利用率，减轻服务器的压力。这都是我们值得去考虑的问题。
常见的三种通信模式
1） 长连接：服务器和客户端的通道一直处于开启状态。合适服务器性能好，客户端数量少的场景。
2） 短连接：只有在发送数据时建立连接，数据发送完后断开连接。一般将数据保存在本地，根据某种逻辑一次性批量提交。适合对实时性不高的应用场景。 3） 特殊长连接：它拥有长连接的特性。当在服务器指定时间内，若没有任何通信，连接就会断开。若客户端再次向服务端发送请求，则需重新建立连接。主要为减小服务端资源占用。 本章重点介特殊长连接。它的设计思想在很多场景中都有，比如QQ的离开状态，电脑的休眠状态。既保证了用户的正常使用，又减轻了服务器的压力。是实际开发中比较常用的通信模式。 它有三个情况：一、服务器和客户端的通道一直处于开启状态。二、指定时间内没有通信则断开连接。三、客户端重新发起请求，则重新建立连接。结合上面的Marshalling 配置工厂，模拟特殊长连接的通信场景。
客户端代码，辅助启动类Bootstrap，配置编解码事件，超时事件，自定义事件。客户端发送请求分两中情况，通信连接时请求和连接断开后请求。package com.itdragon.marshalling;import java.io.File;import java.io.FileInputStream;import java.util.concurrent.TimeUnit;import com.itdragon.utils.ITDragonUtil;import io.netty.bootstrap.Bootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.timeout.ReadTimeoutHandler;public class ITDragonClient { private static final Integer PORT = 8888; private static final String HOST = "127.0.0.1"; private EventLoopGroup group = null; private Bootstrap bootstrap = null; private ChannelFuture future = null; private static class SingletonHolder { static final ITDragonClient instance = new ITDragonClient(); } public static ITDragonClient getInstance(){ return SingletonHolder.instance; } public ITDragonClient() { group = new NioEventLoopGroup(); bootstrap = new Bootstrap(); try { bootstrap.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer () { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(ITDragonMarshallerFactory.buildMarshallingDecoder()); // 配置编码器 socketChannel.pipeline().addLast(ITDragonMarshallerFactory.builMarshallingEncoder()); // 配置解码器 socketChannel.pipeline().addLast(new ReadTimeoutHandler(5)); // 表示5秒内没有连接后断开 socketChannel.pipeline().addLast(new ITDragonClientHandler()); } }) .option(ChannelOption.SO_BACKLOG, 1024); } catch (Exception e) { e.printStackTrace(); } } public void connect(){ try { future = bootstrap.connect(HOST, PORT).sync(); System.out.println("连接远程服务器......"); } catch (Exception e) { e.printStackTrace(); } } public ChannelFuture getChannelFuture(){ if(this.future == null || !this.future.channel().isActive()){ this.connect(); } return this.future; } /** * 特殊长连接： * 1. 服务器和客户端的通道一直处于开启状态， * 2. 在服务器指定时间内，没有任何通信，则断开， * 3. 客户端再次向服务端发送请求则重新建立连接， * 4. 从而减小服务端资源占用压力。 */ public static void main(String[] args) { final ITDragonClient client = ITDragonClient.getInstance(); try { ChannelFuture future = client.getChannelFuture(); // 1. 服务器和客户端的通道一直处于开启状态， for(Long i = 1L; i <= 3L; i++ ){ ITDragonReqData reqData = new ITDragonReqData(); reqData.setId(i); reqData.setName("ITDragon-" + i); reqData.setRequestMsg("NO." + i + " Request"); future.channel().writeAndFlush(reqData); TimeUnit.SECONDS.sleep(2); // 2秒请求一次，服务器是5秒内没有请求则会断开连接 } // 2. 在服务器指定时间内，没有任何通信，则断开， Thread.sleep(6000); // 3. 客户端再次向服务端发送请求则重新建立连接， new Thread(new Runnable() { public void run() { try { System.out.println("唤醒......"); ChannelFuture cf = client.getChannelFuture(); System.out.println("连接是否活跃 : " + cf.channel().isActive()); System.out.println("连接是否打开 : " + cf.channel().isOpen()); ITDragonReqData reqData = new ITDragonReqData(); reqData.setId(4L); reqData.setName("ITDragon-picture"); reqData.setRequestMsg("断开的通道被唤醒了!!!!"); // 路径path自定义 String path = System.getProperty("user.dir") + File.separatorChar + "sources" + File.separatorChar + "itdragon.jpg"; File file = new File(path); FileInputStream inputStream = new FileInputStream(file); byte[] data = new byte[inputStream.available()]; inputStream.read(data); inputStream.close(); reqData.setAttachment(ITDragonUtil.gzip(data)); cf.channel().writeAndFlush(reqData); cf.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } } }).start(); future.channel().closeFuture().sync(); System.out.println("断开连接,主线程结束....."); } catch (Exception e) { e.printStackTrace(); } }} 客户端自定义事务代码，负责将服务器返回的数据打印出来
package com.itdragon.marshalling;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.ReferenceCountUtil;public class ITDragonClientHandler extends ChannelInboundHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("Netty Client active ^^^^^^"); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { ITDragonRespData responseData = (ITDragonRespData) msg; System.out.println("Netty Client : " + responseData.toString()); } catch (Exception e) { e.printStackTrace(); } finally { ReferenceCountUtil.release(msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); }} 服务器代码，辅助启动类ServerBootstrap，配置日志打印事件，编解码事件，超时控制事件，自定义事件。
package com.itdragon.marshalling;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.logging.LogLevel;import io.netty.handler.logging.LoggingHandler;import io.netty.handler.timeout.ReadTimeoutHandler; public class ITDragonServer { private static final Integer PORT = 8888; public static void main(String[] args) { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer () { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(ITDragonMarshallerFactory.buildMarshallingDecoder()); // 配置解码器 socketChannel.pipeline().addLast(ITDragonMarshallerFactory.builMarshallingEncoder()); // 配置编码器 socketChannel.pipeline().addLast(new ReadTimeoutHandler(5)); // 传入的参数单位是秒，表示5秒内没有连接后断开 socketChannel.pipeline().addLast(new ITDragonServerHandler()); } }) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.SO_KEEPALIVE, true); ChannelFuture future = bootstrap.bind(PORT).sync(); future.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { workerGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); } }} 服务器自定义事件代码，负责接收客户端传输的数据，若有附件则下载到receive目录下(这里只是简单的下载逻辑)。
package com.itdragon.marshalling;import java.io.File;import java.io.FileOutputStream;import com.itdragon.utils.ITDragonUtil;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.ReferenceCountUtil; public class ITDragonServerHandler extends ChannelInboundHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("Netty Server active ......"); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { // 获取客户端传来的数据 ITDragonReqData requestData = (ITDragonReqData) msg; System.out.println("Netty Server : " + requestData.toString()); // 处理数据并返回给客户端 ITDragonRespData responseData = new ITDragonRespData(); responseData.setId(requestData.getId()); responseData.setName(requestData.getName() + "-SUCCESS!"); responseData.setResponseMsg(requestData.getRequestMsg() + "-SUCCESS!"); // 如果有附件则保存附件 if (null != requestData.getAttachment()) { byte[] attachment = ITDragonUtil.ungzip(requestData.getAttachment()); String path = System.getProperty("user.dir") + File.separatorChar + "receive" + File.separatorChar + System.currentTimeMillis() + ".jpg"; FileOutputStream outputStream = new FileOutputStream(path); outputStream.write(attachment); outputStream.close(); responseData.setResponseMsg("file upload success , file path is : " + path); } ctx.writeAndFlush(responseData); } catch (Exception e) { e.printStackTrace(); } finally { ReferenceCountUtil.release(msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); }} ITDragonReqData 和 ITDragonRespData 实体类的代码就不贴出来了，github上有源码。
心跳监控案例
在分布式，集群系统架构中，我们需要定时获取各机器的资源使用情况和服务器之间是否保持正常连接状态。以便能在最短的时间内避免和处理问题。类比集群中的哨兵模式。
获取本机数据
可以通过第三方sigar.jar的帮助，获取主机的运行时信息，包括操作系统、CPU使用情况、内存使用情况、硬盘使用情况以及网卡、网络信息。使用很简单，根据自己电脑的系统选择对应的dll文件，然后拷贝到C:\Windows\System32 目录下即可。比如windows7 64位操作系统，则需要sigar-amd64-winnt.dll文件。
下载路径：https://pan.baidu.com/s/1jJSaucI 密码: 48d2ITDragonClient.java，ITDragonCoreParam.java，ITDragonRequestInfo.java，ITDragonServer.java，ITDragonSigarUtil.java，pom.xml 的代码这里就不贴出来了，github上面有完整的源码。
客户端自定义事件代码，负责发送认证信息，定时向服务器发送cpu信息和内存信息。
package com.itdragon.monitoring;import java.net.InetAddress;import java.util.HashMap;import java.util.concurrent.Executors;import java.util.concurrent.ScheduledExecutorService;import java.util.concurrent.ScheduledFuture;import java.util.concurrent.TimeUnit;import org.hyperic.sigar.CpuPerc;import org.hyperic.sigar.Mem;import org.hyperic.sigar.Sigar;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.ReferenceCountUtil; public class ITDragonClientHandler extends ChannelInboundHandlerAdapter{ private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); private ScheduledFuture heartBeat; private InetAddress addr ; //主动向服务器发送认证信息 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("Client 连接一开通就开始验证....."); addr = InetAddress.getLocalHost(); String ip = addr.getHostAddress(); System.out.println("ip : " + ip); String key = ITDragonCoreParam.SALT_KEY.getValue(); // 假装进行了很复杂的加盐加密 // 按照Server端的格式，传递令牌 String auth = ip + "," + key; ctx.writeAndFlush(auth); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { if(msg instanceof String){ String result = (String) msg; if(ITDragonCoreParam.AUTH_SUCCESS.getValue().equals(result)){ // 验证成功，每隔10秒，主动发送心跳消息 this.heartBeat = this.scheduler.scheduleWithFixedDelay(new HeartBeatTask(ctx), 0, 10, TimeUnit.SECONDS); System.out.println(msg); } else { System.out.println(msg); } } } finally { ReferenceCountUtil.release(msg); } } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); if (heartBeat != null) { heartBeat.cancel(true); heartBeat = null; } ctx.fireExceptionCaught(cause); } } class HeartBeatTask implements Runnable { private final ChannelHandlerContext ctx; public HeartBeatTask(final ChannelHandlerContext ctx) { this.ctx = ctx; } public void run() { try { // 采用sigar 获取本机数据，放入实体类中 ITDragonRequestInfo info = new ITDragonRequestInfo(); info.setIp(InetAddress.getLocalHost().getHostAddress()); // ip Sigar sigar = new Sigar(); CpuPerc cpuPerc = sigar.getCpuPerc(); HashMap cpuPercMap = new HashMap (); cpuPercMap.put(ITDragonCoreParam.COMBINED.getValue(), cpuPerc.getCombined()); cpuPercMap.put(ITDragonCoreParam.USER.getValue(), cpuPerc.getUser()); cpuPercMap.put(ITDragonCoreParam.SYS.getValue(), cpuPerc.getSys()); cpuPercMap.put(ITDragonCoreParam.WAIT.getValue(), cpuPerc.getWait()); cpuPercMap.put(ITDragonCoreParam.IDLE.getValue(), cpuPerc.getIdle()); Mem mem = sigar.getMem(); HashMap memoryMap = new HashMap (); memoryMap.put(ITDragonCoreParam.TOTAL.getValue(), mem.getTotal() / 1024L); memoryMap.put(ITDragonCoreParam.USED.getValue(), mem.getUsed() / 1024L); memoryMap.put(ITDragonCoreParam.FREE.getValue(), mem.getFree() / 1024L); info.setCpuPercMap(cpuPercMap); info.setMemoryMap(memoryMap); ctx.writeAndFlush(info); } catch (Exception e) { e.printStackTrace(); } } } 服务器自定义事件代码，负责接收客户端传输的数据，验证令牌是否失效，打印客户端传来的数据。
package com.itdragon.monitoring;import java.util.HashMap;import io.netty.channel.ChannelFutureListener;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;public class ITDragonServerHandler extends ChannelInboundHandlerAdapter { // 令牌验证的map，key为ip地址，value为密钥 private static HashMap authMap = new HashMap (); // 模拟数据库查询 static { authMap.put("xxx.xxx.x.x", "xxx"); authMap.put(ITDragonCoreParam.CLIENT_HOST.getValue(), ITDragonCoreParam.SALT_KEY.getValue()); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("Netty Server Monitoring......."); } // 模拟api请求前的验证 private boolean auth(ChannelHandlerContext ctx, Object msg) { System.out.println("令牌验证..............."); String[] ret = ((String) msg).split(","); String clientIp = ret[0]; // 客户端ip地址 String saltKey = ret[1]; // 数据库保存的客户端密钥 String auth = authMap.get(clientIp); // 客户端传来的密钥 if (null != auth && auth.equals(saltKey)) { ctx.writeAndFlush(ITDragonCoreParam.AUTH_SUCCESS.getValue()); return true; } else { ctx.writeAndFlush(ITDragonCoreParam.AUTH_ERROR.getValue()).addListener(ChannelFutureListener.CLOSE); return false; } } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // 如果传来的消息是字符串，则先验证 if (msg instanceof String) { auth(ctx, msg); } else if (msg instanceof ITDragonRequestInfo) { ITDragonRequestInfo info = (ITDragonRequestInfo) msg; System.out.println("--------------------------------------------"); System.out.println("当前主机ip为: " + info.getIp()); HashMap cpu = info.getCpuPercMap(); System.out.println("cpu 总使用率: " + cpu.get(ITDragonCoreParam.COMBINED.getValue())); System.out.println("cpu 用户使用率: " + cpu.get(ITDragonCoreParam.USER.getValue())); System.out.println("cpu 系统使用率: " + cpu.get(ITDragonCoreParam.SYS.getValue())); System.out.println("cpu 等待率: " + cpu.get(ITDragonCoreParam.WAIT.getValue())); System.out.println("cpu 空闲率: " + cpu.get(ITDragonCoreParam.IDLE.getValue())); HashMap memory = info.getMemoryMap(); System.out.println("内存总量: " + memory.get(ITDragonCoreParam.TOTAL.getValue())); System.out.println("当前内存使用量: " + memory.get(ITDragonCoreParam.USED.getValue())); System.out.println("当前内存剩余量: " + memory.get(ITDragonCoreParam.FREE.getValue())); System.out.println("--------------------------------------------"); ctx.writeAndFlush("info received!"); } else { ctx.writeAndFlush("connect failure!").addListener(ChannelFutureListener.CLOSE); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); }} 总结
1 Netty的编解码功能很好的解决了Java序列化 无法跨语言，序列化后码流太大，序列化性能太低等问题
2 JBoss的Marshalling是一个Java对象的序列化API包，修正了JDK自带的序列化包的很多问题，又兼容java.io.Serializable接口，缺点使用范围小。 3 特殊长连接可以减小服务端资源占用压力，是一种比较常用的数据通信方式。 4 Netty可以用做心跳监测，定时获取被监听机器的数据信息。推荐文档
Netty 能做什么？学Netty有什么用？
https://www.zhihu.com/question/24322387 http://blog.csdn.net/broadview2006/article/details/46041995 Marshalling : http://jbossmarshalling.jboss.org/Netty 编解码数据通信和心跳监控案例到这里就结束了，感谢大家的阅读，欢迎点评。如果你觉得不错，可以"推荐"一下。也可以"关注"我，获得更多丰富的知识。