public class StatisticsServer {
  //每次發(fā)送接收的數據包大小
  private final int MAX_BUFF_SIZE = 1024 * 10;
  //服務端監(jiān)聽端口，客戶端也通過該端口發(fā)送數據
  private int port;
  private DatagramChannel channel;
  private Selector selector;
  private ScheduledExecutorService es = Executors.newScheduledThreadPool(1);
  public void init() throws IOException {
    //創(chuàng)建通道和選擇器
    selector = Selector.open();
    channel = DatagramChannel.open();
    //設置為非阻塞模式
    channel.configureBlocking(false);
    channel.socket().bind(new InetSocketAddress(port));
    //將通道注冊至selector，監(jiān)聽只讀消息（此時服務端只能讀數據，無法寫數據）
    channel.register(selector, SelectionKey.OP_READ);
    //使用線程的方式，保證服務端持續(xù)等待接收客戶端數據
    es.scheduleWithFixedDelay(new Runnable() {
      @Override
      public void run() {
        try {
          while(selector.select() > 0) {
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while(iterator.hasNext()) {
              SelectionKey key = iterator.next();
              try {
                iterator.remove();
                if(key.isReadable()) {
                  //接收數據
                  doReceive(key);
                }
              } catch (Exception e) {
                logger.error("SelectionKey receive exception", e);
                try {
                  if (key != null) {
                    key.cancel();
                    key.channel().close();
                  }
                } catch (ClosedChannelException cex) {
                  logger.error("Close channel exception", cex);
                }
              }
            }
          }
        } catch (IOException e) {
          logger.error("selector.select exception", e);
        }
      }
    }, 0L, 2L, TimeUnit.MINUTES);
  }
  //處理接收到的數據
  private void doReceive(SelectionKey key) throws IOException {
    String content = "";
    DatagramChannel sc = (DatagramChannel) key.channel();
    ByteBuffer buffer = ByteBuffer.allocate(MAX_BUFF_SIZE);
    buffer.clear();
    sc.receive(buffer);
    buffer.flip();
    while(buffer.hasRemaining()) {
      byte[] buf = new byte[buffer.limit()];
      buffer.get(buf);
      content += new String(buf);
    }
    buffer.clear();
    logger.debug("receive content="+content);
    if(StringUtils.isNotBlank(content)) {
      doSave(content);
    }
  }
 }

DatagramChannel channel = DatagramChannel.open();

    StringBuilder sb = new StringBuilder();
    sb.append("2017-03-09 12:30:00;")
        .append("aaa")
        .append("testapp;")
        .append("test.do;")
        .append("param=hello;")
        .append("test;")
        .append("100;")
        .append("1");
    ByteBuffer buffer = ByteBuffer.allocate(10240);
    buffer.clear();
    buffer.put(sb.toString().getBytes());
    buffer.flip();
    //此處IP為服務端IP地址，端口和服務端的端口一致
    int n = channel.send(buffer, new InetSocketAddress("127.0.0.1", 8080));
    System.out.println(n);
    //每次數據發(fā)送完畢之后，一定要調用close方法，來關閉占用的udp端口，否則程序不結束，端口不會釋放
    channel.close();