• Kafka中Zero-Copy的相关总结

    简介

    Kafka用到了零拷贝（Zero-Copy）技术来提升性能。所谓的零拷贝是指数据直接从磁盘复制文件到网卡设备，而无需经过应用程序，减少了内核和用户空间之间的上下文切换。

    下面这个过程是不采用零拷贝技术时，从磁盘中读取文件然后通过网卡发送出去的流程，可以看到：经历了 4 次拷贝，4 次上下文切换。

    如果采用零拷贝技术（底层通过 sendfile64 方法实现），流程将变成下面这样。可以看到：只需 3 次拷贝以及 2 次上下文切换，显然性能更高。

    Kafka源码

    在Kafka的源码中，使用zero-copy相关的代码如下：

    org.apache.kafka.common.record.FileRecords.java

    1.     @Override
    2.     public long writeTo(TransferableChannel destChannel, long offset, int length) throws IOException {
    3.         long newSize = Math.min(channel.size(), end) - start;
    4.         int oldSize = sizeInBytes();
    5.         if (newSize < oldSize)
    6.             throw new KafkaException(String.format(
    7.                     "Size of FileRecords %s has been truncated during write: old size %d, new size %d",
    8.                     file.getAbsolutePath(), oldSize, newSize));
    9.  
    10.         long position = start + offset;
    11.         long count = Math.min(length, oldSize - offset);
    12.         return destChannel.transferFrom(channel, position, count);
    13.     }

    进一步跟踪代码到 org.apache.kafka.common.network.PlaintextTransportLayer.java，其调用的方法是 fileChannel.transferTo(position, count, socketChannel);，这也是从写代码的角度最需要关注的地方，即，通过文件的channel，直接调用其transferTo方法，传入参数为文件position, count，以及SocketChannel对象。这样就能直接将文件内容发送到SocketChannel对象所绑定的网卡，而不需要将文件的内容从内核空间复制到用户空间，然后再从用户空间复制到网卡设备的内核空间，减少了上下文切换和数据的复制，从而提高性能。

    1. @Override
    2.     public long transferFrom(FileChannel fileChannel, long position, long count) throws IOException {
    3.         return fileChannel.transferTo(position, count, socketChannel);
    4.     }

    进一步跟踪 transferTo 方法，可以发现，其最终在Linux上会调用到 sun.nio.ch.FileChannelImpl.java 的 transferTo0方法，这是一个native方法，底层实现是调用sendfile64()的C函数，这与通过NIO快速拷贝文件非常类似，参考：Kafka中mmap的相关总结

    1. private long transferToDirectlyInternal(long position, int icount,
    2.                                             WritableByteChannel target,
    3.                                             FileDescriptor targetFD)
    4.         throws IOException
    5.     {
    6.         assert !nd.transferToDirectlyNeedsPositionLock() ||
    7.                Thread.holdsLock(positionLock);
    8.  
    9.         long n = -1;
    10.         int ti = -1;
    11.         try {
    12.             beginBlocking();
    13.             ti = threads.add();
    14.             if (!isOpen())
    15.                 return -1;
    16.             do {
    17.                 n = transferTo0(fd, position, icount, targetFD);
    18.             } while ((n == IOStatus.INTERRUPTED) && isOpen());
    19.             if (n == IOStatus.UNSUPPORTED_CASE) {
    20.                 if (target instanceof SinkChannelImpl)
    21.                     pipeSupported = false;
    22.                 if (target instanceof FileChannelImpl)
    23.                     fileSupported = false;
    24.                 return IOStatus.UNSUPPORTED_CASE;
    25.             }
    26.             if (n == IOStatus.UNSUPPORTED) {
    27.                 // Don't bother trying again
    28.                 transferSupported = false;
    29.                 return IOStatus.UNSUPPORTED;
    30.             }
    31.             return IOStatus.normalize(n);
    32.         } finally {
    33.             threads.remove(ti);
    34.             end (n > -1);
    35.         }
    36.     }

    Java实例

    下面我们通过一个简单的Java实例来看看如何使用zero-copy技术来直接向socketchannel发送文件内容。

    Server

    1. package com.my.kafka.zerocopy;
    2.  
    3. import java.io.IOException;
    4. import java.net.InetSocketAddress;
    5. import java.nio.ByteBuffer;
    6. import java.nio.channels.SelectionKey;
    7. import java.nio.channels.Selector;
    8. import java.nio.channels.ServerSocketChannel;
    9. import java.nio.channels.SocketChannel;
    10. import java.util.Iterator;
    11. import java.util.Set;
    12.  
    13. public class Server {
    14. 	private static final int PORT = 8888;
    15. 	private Selector selector;
    16.  
    17. 	private void startServer() {
    18. 		try {
    19. 			selector = Selector.open();
    20. 			ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    21. 			serverSocketChannel.configureBlocking(false);
    22. 			serverSocketChannel.bind(new InetSocketAddress(PORT));
    23. 			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    24.  
    25. 			System.out.println("SERVER: Server started on port: " + PORT);
    26. 			while (true) {
    27. 				int selectCode = selector.select(1000);
    28. 				if (selectCode == 0) {
    29. 					continue;
    30. 				}
    31. 				Set<SelectionKey> keys = selector.selectedKeys();
    32. 				Iterator<SelectionKey> it = keys.iterator();
    33. 				while (it.hasNext()) {
    34. 					SelectionKey key = it.next();
    35. 					it.remove();
    36. 					if (key.isAcceptable()) {
    37. 						System.out.println("SERVER: Received a request.");
    38. 						serverSocketChannel = (ServerSocketChannel) key.channel();
    39. 						SocketChannel socketChannel = serverSocketChannel.accept();
    40. 						socketChannel.configureBlocking(false);
    41. 						socketChannel.register(selector, SelectionKey.OP_READ);
    42. 					} else if (key.isReadable()) {
    43. 						readData(key);
    44. 					} else {
    45. 						System.out.println("SERVER: Key is no recognised, key: " + key);
    46. 					}
    47. 				}
    48. 			}
    49. 		} catch (IOException e) {
    50. 			e.printStackTrace();
    51. 		}
    52. 	}
    53.  
    54. 	private void readData(SelectionKey key) {
    55. 		SocketChannel socketChannel = (SocketChannel) key.channel();
    56. 		ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
    57. 		try {
    58. 			socketChannel.read(buffer);
    59. 			buffer.flip();
    60. 			byte[] bytes = new byte[buffer.limit()];
    61. 			buffer.get(bytes);
    62. 			String message = new String(bytes).toString();
    63. 			System.out.println("SERVER: Received message from client: " + message);
    64. 			socketChannel.write(ByteBuffer.wrap("This is response from server side.".getBytes()));
    65. 		} catch (Exception e) {
    66. 			e.printStackTrace();
    67. 		}
    68. 	}
    69.  
    70. 	public static void main(String[] args) {
    71. 		new Server().startServer();
    72. 	}
    73. }

    Client

    1. package com.my.kafka.zerocopy;
    2.  
    3. import java.io.FileNotFoundException;
    4. import java.io.IOException;
    5. import java.io.RandomAccessFile;
    6. import java.net.InetSocketAddress;
    7. import java.nio.ByteBuffer;
    8. import java.nio.channels.FileChannel;
    9. import java.nio.channels.SelectionKey;
    10. import java.nio.channels.Selector;
    11. import java.nio.channels.SocketChannel;
    12. import java.util.Iterator;
    13.  
    14. public class Client {
    15.  
    16. 	private Selector selector;
    17.  
    18. 	private static String FILE_PATH = "C:\\temp\\test_channel.txt";
    19.  
    20. 	public void initClient(String ip, int port) throws IOException {
    21. 		SocketChannel channel = SocketChannel.open();
    22. 		channel.configureBlocking(false);
    23. 		this.selector = Selector.open();
    24. 		channel.connect(new InetSocketAddress(ip, port));
    25. 		channel.register(selector, SelectionKey.OP_CONNECT);
    26. 	}
    27.  
    28. 	public void listen() throws IOException {
    29. 		while (true) {
    30. 			int selectCode = selector.select(1000);
    31. 			if (selectCode == 0) {
    32. 				continue;
    33. 			}
    34.  
    35. 			Iterator<SelectionKey> it = this.selector.selectedKeys().iterator();
    36. 			while (it.hasNext()) {
    37. 				SelectionKey key = (SelectionKey) it.next();
    38. 				if (key.isConnectable()) {
    39. 					SocketChannel channel = (SocketChannel) key.channel();
    40. 					if (channel.isConnectionPending()) {
    41. 						channel.finishConnect();
    42. 					}
    43. 					channel.configureBlocking(false);
    44. 					channel.register(this.selector, SelectionKey.OP_READ);
    45. 					sendFileContent(channel);
    46. 				} else if (key.isReadable()) {
    47. 					readData(key);
    48. 				}
    49. 				it.remove();
    50. 			}
    51. 		}
    52. 	}
    53.  
    54. 	/*
    55. 	 * This is key method to send file based on zero-copy
    56. 	 * Just call method channel.transferTo(0, channel.size(), socketChannel);
    57. 	 */
    58. 	private static void sendFileContent(SocketChannel socketChannel) {
    59. 		try (RandomAccessFile raf = new RandomAccessFile(FILE_PATH, "rw"); FileChannel channel = raf.getChannel()) {
    60. 			channel.transferTo(0, channel.size(), socketChannel);
    61. 		} catch (IOException e) {
    62. 			e.printStackTrace();
    63. 		}
    64. 	}
    65.  
    66. 	private static void prepareFile() {
    67. 		try (RandomAccessFile raf = new RandomAccessFile(FILE_PATH, "rw"); FileChannel channel = raf.getChannel()) {
    68. 			String str = "This is file content from client.";
    69. 			ByteBuffer buffer = ByteBuffer.allocate(str.getBytes().length);
    70. 			buffer.clear();
    71. 			buffer.put(str.getBytes());
    72. 			buffer.flip(); // Change mode to read, invoke this method when reading data from buffer
    73. 			channel.write(buffer);
    74. 		} catch (FileNotFoundException e) {
    75. 			e.printStackTrace();
    76. 		} catch (IOException e1) {
    77. 			e1.printStackTrace();
    78. 		}
    79. 	}
    80.  
    81. 	public void readData(SelectionKey key) throws IOException {
    82. 		SocketChannel channel = (SocketChannel) key.channel();
    83. 		ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
    84. 		channel.read(buffer);
    85. 		byte[] data = buffer.array();
    86. 		String msg = new String(data).trim();
    87. 		System.out.println("CLIENT: Received message from server: " + msg);
    88. 	}
    89.  
    90. 	public static void main(String[] args) throws Exception {
    91. 		prepareFile();
    92. 		Client client = new Client();
    93. 		client.initClient("localhost", 8888);
    94. 		client.listen();
    95. 	}
    96. }

    运行结果

  • 相关阅读:
    华为OD机试真题-剩余银饰的重量-2024年OD统一考试（C卷D卷）
    环路与快速破环
    Python所有常见功能大汇总
    抖音短视频实操：矩阵号之为什么要做矩阵号和如何做矩阵号（下）
    Mac brew 安装与使用
    Pytorch入门实例
    openGauss学习笔记-130 openGauss 数据库管理-参数设置-重设参数
    Linux Vim 进阶教程
    解决跨域问题的FastAPI应用及常见报错解析
    Kafka系列之：Kafka Connect错误报告设置
  • 原文地址：https://blog.csdn.net/funnyrand/article/details/125513774