Netty Example - server echo user input message


Reference
Maven
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.10.Final</version>
</dependency>
Description
這個例子跟 guide 上的不太相同, 除了原本的 echo 行為.
另外有做讓人 input message, 從 client 送給 server 做 echo.
MainClass
public class EchoServerMain {

    public static void main(String[] params) throws Exception {
        new Thread(() -> {
            try {
                new EchoServer(1234).run();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        TimeUnit.SECONDS.sleep(1);
        new EchoClient("localhost", 1234).connectRunAndExit();
    }

    public static class EchoClient {

        private final String ip;
        private final int port;

        EchoClient(String ip, int port) {
            this.port = port;
            this.ip = ip;
        }

        public void connectRunAndExit() throws InterruptedException {
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(workerGroup);
                b.channel(NioSocketChannel.class);
                b.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new EchoClientHandler());
                    }
                });

                ChannelFuture f = b.connect(ip, port).sync();

                Channel ch = f.channel();
                System.out.println("scan...");
                Scanner scanner = new Scanner(System.in);
                while (scanner.hasNextLine()) {
                    String msg = scanner.nextLine();
                    System.out.println("Input:" + msg);
                    ByteBuf msgBuffer = Unpooled.wrappedBuffer(msg.getBytes(CharsetUtil.UTF_8));
                    ch.writeAndFlush(msgBuffer);
                }


                // Wait until the connection is closed.
                f.channel().closeFuture().sync();
            } finally {
                workerGroup.shutdownGracefully().sync();
            }
        }
    }


    public static class EchoServer {
        private int port;

        public EchoServer(int port) {
            this.port = port;
        }

        public void run() throws Exception {
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(group)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                socketChannel.pipeline().addLast(new EchoServerHandler());
                            }
                        });

                ChannelFuture f = b.bind(port).sync();
                System.out.println("bind done");
                f.channel().closeFuture().sync();
                System.out.println("close done");
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    }


}
EchoClientHandler
public class EchoClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        System.out.println("Client receive:" + ((ByteBuf)msg).toString(Charset.defaultCharset()));
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
    }
}

EchoServerHandler
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        System.out.println("triggered:" + evt);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server receive:" + msg);
        ctx.writeAndFlush(msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("connection active:" + ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("connection inactive:" + ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

Output

RabbitMQ Example

Reference

Example

一個 thread 持續收, main thread 持續送,
在 4 core PC 上每秒約5000+ 個訊息
package amqp;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author <a href="https://github.com/shooeugenesea">isaac</a>
 */
public class ProducerConsumerMain {

    private final static AtomicInteger sentMsg = new AtomicInteger();
    private final static AtomicInteger receiveMsg = new AtomicInteger();
    private final static String MSG_5K;
    private final static String QUEUE_NAME = "hello";

    static {
        StringBuilder sb = new StringBuilder();
        for ( int i = 0; i < 5000; i++ ) {
            sb.append(String.valueOf(i));
        }
        MSG_5K = sb.toString();
    }

    public static void main(String[] params) throws IOException, TimeoutException {
        new Thread(){
            @Override
            public void run() {
                try {
                    ConnectionFactory factory = new ConnectionFactory();
                    factory.setHost("localhost");
                    Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel();

                    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

                    Consumer consumer = new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope,
                                                   AMQP.BasicProperties properties, byte[] body)
                                throws IOException {
                            String message = new String(body, "UTF-8");

                            receiveMsg.incrementAndGet();
//                            System.out.println(" [x] Received message '" + message.length() + "'");
                        }
                    };
                    channel.basicConsume(QUEUE_NAME, true, consumer);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }.start();

        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(new Runnable(){

            @Override
            public void run() {
                System.out.println("receiveMsg:" + receiveMsg.getAndSet(0) + ", sentMsg:" + sentMsg.getAndSet(0));
            }
        }, 0, 1, TimeUnit.SECONDS);

        send();
    }

    private static void send() {
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            String message = MSG_5K;
            for ( ;; ) {
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                sentMsg.incrementAndGet();
//            System.out.println(" [x] Sent '" + message + "'");
            }

//            channel.close();
//            connection.close();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

}

Google Protobuf - Basic Example

Reference

Example

Install protobuf

  1. Download from Git: https://github.com/protocolbuffers/protobuf
  2. Install protobuf
  3. Setup protobuf folder to PATH
  4. Try command: protoc

Define proto

syntax = "proto2";

package examples;

option java_package = "examples";
option java_outer_classname = "AddressBookProtos";

message Person {
    required string name = 1;
    required int32 id = 2;
    optional string email = 3;

    enum PhoneType {
        MOBILE = 0;
        HOME = 1;
        WORK = 2;
    }

    message PhoneNumber {
        required string number = 1;
        optional PhoneType type = 2 [default = HOME];
    }

    repeated PhoneNumber phones = 4;
}

message AddressBook {
    repeated Person people = 1;
}

Generate Java Code

Execute command: 

protoc -I /Users/liaoisaac/projects/study-practice/src/main/resources -java_out=/Users/liaoisaac/projects/study-practice/src/main/java/ /Users/liaoisaac/projects/study-practice/src/main/resources/addressbook.proto
Java code will be generated in "/Users/liaoisaac/projects/study-practice/src/main/java/"

Example with Maven Plugin

Introduce Maven

Define Maven plugin: protobuf-maven-plugin
(注意我把 output dir 改到我想要的 generated-java folder 了)



<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <groupId>shooeugenesea</groupId>
 <artifactId>study-examples</artifactId>
 <version>0.0.1-SNAPSHOT</version>
 <dependencies>
  <dependency>
   <groupId>junit</groupId>
   <artifactId>junit</artifactId>
   <version>4.11</version>
   <scope>test</scope>
  </dependency>
  <dependency>
   <groupId>com.google.protobuf</groupId>
   <artifactId>protobuf-java</artifactId>
   <version>3.3.0</version>
  </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.7</version>
        </dependency>
    </dependencies>
 <build>
  <plugins>
   <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-compiler-plugin</artifactId>
    <version>3.7.0</version>
    <configuration>
     <source>1.8</source>
     <target>1.8</target>
    </configuration>
   </plugin>
   <plugin>
    <groupId>org.xolstice.maven.plugins</groupId>
    <artifactId>protobuf-maven-plugin</artifactId>
    <version>0.5.1</version>
    <configuration>
     <protocExecutable>/usr/local/bin/protoc</protocExecutable>
                    <outputDirectory>${basedir}/src/main/generated-java</outputDirectory>
    </configuration>
    <executions>
     <execution>
      <goals>
       <goal>compile</goal>
       <goal>test-compile</goal>
      </goals>
     </execution>
    </executions>
   </plugin>
   <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-compiler-plugin</artifactId>
    <version>3.1</version>
    <configuration>
     <verbose>true</verbose>
     <fork>true</fork>
     <compilerVersion>1.8</compilerVersion>
    </configuration>
   </plugin>
   <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-compiler-plugin</artifactId>
    <version>3.3</version>
    <configuration>
     <source>1.8</source>
     <target>1.8</target>
    </configuration>
   </plugin>
  </plugins>
 </build>
</project>
設定好之後執行 mvn clean package 就可以根據 proto generate 出 java code

Read Write generated objects

package examples;

import java.io.*;

public class ReadWriteProtobufMain {

    public static void main(String[] params) throws IOException {
        AddressBookProtos.Person person = AddressBookProtos.Person.newBuilder()
                .setEmail("email@com")
                .setName("myname")
                .setId(123)
                .addPhones(AddressBookProtos.Person.PhoneNumber.newBuilder()
                        .setType(AddressBookProtos.Person.PhoneType.HOME)
                        .setNumber("12345678")
                        .build())
                .build();


        try (PipedOutputStream out = new PipedOutputStream();
             PipedInputStream in = new PipedInputStream(out)) {
            System.out.println("write Person:\n" + person);
            person.writeTo(out);
            out.close();

            AddressBookProtos.Person readPerson = AddressBookProtos.Person.newBuilder().mergeFrom(in).build();
            System.out.println("read Person:\n" + readPerson);
        }

    }

}

Output


Spring Integration - Basic Terms

Reference

Terms

org.springframework.integration.Message<T>

用來封裝訊息, 成員包含 MessageHeader 與 Payload

org.springframework.integration.MessageHeaders

MessageHeaders 是 immutable 的, 包含一些預設的 attribute: id, timestamp, correlation id, and priority

Payloads

用來裝 message body, 可以自訂 transformer 來傳遞封包

Message Channel

Message Channel 用來傳遞封包, 也用來 decouple producer 與 consumer.
Message Channel 有兩種模式: 
  1. point to point, 一個封包只會被一個 consumer 收到
  2. public/subscribe: 一個封包會被多個 consumer 收到

Message Endpoints

endpoint 泛指 Spring Integration 裡面的各種 component.
  • Message Adapter: 資料可從外部系統透過 adapter 送進 Spring Integration
  • Transformer: 轉換訊息
  • Filter: 決定 Message 是否傳給 Message Channel
  • Router: 透過 Message 的內容判斷要送給哪一個 Message Channel
  • Splitter: 把一個 Message 切成好幾個轉送給不同適合的 Message Channel
  • Aggregator: 把多個訊息合併成一個
  • Service activator: Message Channel 與 Service instance 之間的介面

Example

spring-context.xml


<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-5.0.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
        <context:component-scan base-package="examples" />
        <int:channel id="input"/>
        <int:channel id="output">
            <int:queue capacity="10"/>
        </int:channel>
        <int:service-activator input-channel="input"
                           output-channel="output"
                           ref="messageHandler"/>
</beans>

examples/MessageHandler.java

@Component
public class MessageHandler {
    @ServiceActivator
    public String handleMessage(String message) {
        System.out.println("Received message: " + message);
        return "MESSAGE:" + message;
    }
}

examples/Application.java

public class Application {
    public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring-context.xml");
        context.start();

        MessageChannel input = context.getBean("input", MessageChannel.class );
        PollableChannel output = context.getBean("output", PollableChannel.class );

        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()) {
            input.send(MessageBuilder.withPayload(scanner.nextLine()).build());
        }
    }
}

Run, input message and send to MessageHandler to print


Off Heap Cache - OHCache - Simple Example

Reference

What

OHC 是一個 off heap 的 cache library

Why

如果是用 map 做 cache, 資料都在 heap 裡面, heap 裡面的問題就是
  1. 會被 GC, GC 就會週期性的對 JVM 效能產生影響
  2. 如果要 persist 需要另外實作
如果是放在 off heap, 比方說用 mmap, 或是用 MappedByteBuffer access 的記憶體, 就是放在 heap memory 之外, 由作業系統控制著. 
好處是記憶體量不用被 heap size 限制, 缺點是要自己小心控制, 不然會造成系統問題.'

How

Maven

<dependency>
 <groupId>org.caffinitas.ohc</groupId>
 <artifactId>ohc-core</artifactId>
 <version>0.4.4</version>
</dependency>

Codes

public class SimpleMain {

    public static void main(String[] params) {
        Scanner scanner = new Scanner(System.in);
        while(scanner.hasNextLine()) {
            if (scanner.nextLine().trim().equals("GO")) {
                System.out.println(Runtime.getRuntime().freeMemory()/1024/1024 + "/" + Runtime.getRuntime().totalMemory()/1024/1024);
                OHCache<String,String> c = OHCacheBuilder.<String,String>newBuilder()
                        .keySerializer(new StringSerializer())
                        .valueSerializer(new StringSerializer())
                        .build();
                c.put("A", "A");
                System.out.println(c.get("A"));
                IntStream.range(0, 1000_000).forEach(idx -> {
                    c.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
                });
                System.out.println(Runtime.getRuntime().freeMemory()/1024/1024 + "/" + Runtime.getRuntime().totalMemory()/1024/1024);
            } else {
                System.out.println("input 'GO' then click enter");
            }
        }
    }

    private static class StringSerializer implements CacheSerializer<String> {

        @Override
        public void serialize(String s, ByteBuffer byteBuffer) {
            byteBuffer.put(s.getBytes());
        }

        @Override
        public String deserialize(ByteBuffer byteBuffer) {
            return StandardCharsets.UTF_8.decode(byteBuffer).toString();
        }

        @Override
        public int serializedSize(String s) {
            return s.getBytes().length;
        }
    }

}

Output


這張是輸入了好幾次 GO, 每次都會輸入一百萬次 key value 到 cache.

jconsole 與系統資源使用圖可以看到 heap 只用不到 100MB, 但實際上已經用了 600MB.
就是都放在 off heap 中

ForkJoinPool - Thread Management

Reference

Question

原本以為 new ForkJoinPool(2) 像這樣的宣告, 是讓 ForkJoinPool 最多維持 2 個 thread 來執行.
可是當實際觀察的時候, 卻發現有非常多個 thread 被叫起來跑.
比方說原本 fork -> join 的範例, 如果把 thread name 印出來, 就會發現有很多個 thread 被叫起來.
而且每秒印出的 pool 也會發現 pool size 變大.

Codes: Thread size exceed parallel level

public class ForkJoinPoolManyThreadMain {

    static final CountDownLatch latch = new CountDownLatch(1);
    private static final AtomicInteger threadCount = new AtomicInteger();

    public static void main(String[] params) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> {
            System.out.println(pool); }, 0, 1, TimeUnit.SECONDS);
        pool.execute(new MainAction());
        latch.await();
        s.shutdown();
        pool.shutdown();
    }


    private static class MainAction extends RecursiveAction {

        @Override
        protected void compute() {
            List<SubAction> actions = IntStream.range(0,100).mapToObj(idx -> new SubAction()).collect(toList());
            actions.forEach(SubAction::fork);
            actions.forEach(SubAction::join);
            latch.countDown();
        }
    }

    private static class SubAction extends RecursiveAction {

        @Override
        protected void compute() {
            try {
                System.out.println(Thread.currentThread().getName() + " sleep 1 seconds");
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


}

Output

從印出來的 log 就可以發現: thread 超過 2 個到達 14 個. pool size 實際上也到達 16





Requirement

發現這件事情的時候覺得無法理解, 同時也發現原本以為 main task 的 thread 是跟被 join 的 thread 合再一起了. 結果看起來是有很多新的 thread 被產生出來執行. 如此我們怎麼確保不會有過多的 thread 被建立出來? 

Suggestion: Customized ThreadFactory? (Don't do this)

因為 ForkJoinPool 裡的這段程式, 網路上有人建議是提供自己的 thread factory, 如果超過指定的 thread 數量就回傳 null. 如此就被 ForkJoinPool 當成一個錯誤而繼續使用既有的 thread.

這個做法的確可以控制建立的 thread 數量 , 不過回傳 null 對 ForkJoinPool 來說是一個錯誤, 以 ForkJoinPool 的規則來說這個時候是要 createWorker 的, 不應該出現錯誤而回傳 null.



How: Join after done

突然想到會產生新的 thread 是不是因為 join 導致 MainAction 停住, 但又不是正在執行, 所以為了避免 starvation 使 ForkJoinPool 需要建立一個新的 thread. 所以試著先判斷 isDone, done 才 join. 結果就可以了.  (但還不清楚原因)

Working Code

public class ForkJoinPoolFixedThreadMain {

    public static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        MainAction action = new MainAction();
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> System.out.println(pool), 0, 1, TimeUnit.SECONDS);
        pool.execute(action);
        latch.await();
        s.shutdown();
    }

    private static class MainAction extends RecursiveAction {

        @Override
        protected void compute() {
            List<SubAction> actions = IntStream.range(0,100).mapToObj(idx -> new SubAction()).collect(toList());
            actions.forEach(SubAction::fork);
            while(!actions.isEmpty()) {
                for (Iterator<SubAction> i = actions.iterator(); i.hasNext(); ) {
                    SubAction action = i.next();
                    if (action.isDone()) {
                        action.join();
                        i.remove();
                    }
                }
            }
            latch.countDown();
        }
    }

    private static class SubAction extends RecursiveAction {

        @Override
        protected void compute() {
            long start = System.currentTimeMillis();
            while (true) {
                if (System.currentTimeMillis() - start > 1000) {
                    break;
                }
            }
        }
    }


}

Output

因為呼叫 new ForkJoinPool() 所以 pool size 最多就維持在 8 (我的電腦有 8 core)

ForkJoinPool - Error Handling

Reference

Error Handling

如果 fork 出去的 task/action 有 exception, 呼叫 join 的時候要注意 catch exception.
不然 task/action 就死掉失控了.
而且可怕的是: 不會有 log.

Codes

這段程式是說每個 action 有個 id, 奇數 id 會報 exception.

Exception without catch when join

先是不要 catch 的情況
public class ForkJoinPoolErrorMain {

    static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> System.out.println(pool), 0, 1, TimeUnit.SECONDS);
        pool.submit(new ErrorAction(MAIN_TASK_ID));
        latch.await(10, TimeUnit.SECONDS);
        pool.shutdown();
        s.shutdown();
    }

    static class ErrorAction extends RecursiveAction {

        public static final int MAIN_TASK_ID = -1;
        private final int id;

        ErrorAction(int id) {
            this.id = id;
        }

        private boolean isSubTask() {
            return id >= 0;
        }

        @Override
        protected void compute() {
            if (isSubTask()) {
                if (id % 2 == 1) {
                    throw new IllegalStateException("Error when id is odd number");
                }
            } else {
                List<ErrorAction> actions = IntStream.range(0,10).mapToObj(idx -> new ErrorAction(idx)).collect(Collectors.toList());
                actions.forEach(ErrorAction::fork);
                actions.forEach(action -> {
                    action.join();
                });
                latch.countDown();
            }
            System.out.println("compute done. id=" + id);
        }
    }

}
輸出如下, 不過一個 exception 也沒有.
main action 也沒有完成.(main action id = -1)











Exception with catch statement

有 catch 的程式如下
public class ForkJoinPoolErrorMain {

    static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> System.out.println(pool), 0, 1, TimeUnit.SECONDS);
        pool.submit(new ErrorAction(MAIN_TASK_ID));
        latch.await(10, TimeUnit.SECONDS);
        pool.shutdown();
        s.shutdown();
    }

    static class ErrorAction extends RecursiveAction {

        public static final int MAIN_TASK_ID = -1;
        private final int id;

        ErrorAction(int id) {
            this.id = id;
        }

        private boolean isSubTask() {
            return id >= 0;
        }

        @Override
        protected void compute() {
            if (isSubTask()) {
                if (id % 2 == 1) {
                    throw new IllegalStateException("Error when id is odd number");
                }
            } else {
                List<ErrorAction> actions = IntStream.range(0,10).mapToObj(idx -> new ErrorAction(idx)).collect(Collectors.toList());
                actions.forEach(ErrorAction::fork);
                actions.forEach(action -> {
                    try {
                        action.join();
                    } catch (Exception ex) {
                        System.err.println(ex + ", id=" + action.id);
                    }
                });
                latch.countDown();
            }
            System.out.println("compute done. id=" + id);
        }
    }

}

這個程式就可以印 error, 而且 maintask 可以執行完成











Exception with ExecutorService

如果是 ExecutorService, 就算是丟 RuntimeException.
ExecutorService 還是會把 exception 印到 console 上.
public class ExecutorServiceWithExceptionMain {

    public static void main(String[] params) {
        ExecutorService ex = Executors.newSingleThreadExecutor();
        ex.execute(() -> {
            throw new IllegalStateException("test throw illegalState");
        });
        ex.shutdown();
    }


}
即使是直接丟 exception 也會印到 console.

ForkJoinPool - Work stealing


Reference

前篇: Java Concurrency - ForkJoinPool 的 deadlock
下篇: ForkJoinPool - Error Handling

Introduction

ForkJoinPool 實作 work stealing 的概念, 簡單說明一下

Work-stealing

  1. 每個 thread 會有一個 queue, queue 裡面放的是 CPU bound 的工作
  2. queue 裡面的工作可能會產生新的能平行作業的工作
  3. 新產生的工作會放在 queue 裡面
  4. 如果有 thread 把 queue 裡面的工作做完了, 就會去別的 queue 拿工作來處理 (steal)

Java Concurrency - ForkJoinPool 的 deadlock

Reference

Will ForkJoinTask encounter deadlock problem?

在看 ForkJoinTask 的時候, 看到 ForkJoinTask 可以先 fork 去執行, 再來呼叫 join 等待結束.
如圖, 些動作總共需要幾個 thread?
原本想:
1. MainTask 一個 thread, 兩個 subtask 各一個 thread 執行.
2. 當兩個 subtask 要 join 的時候, MainTask 會等待
如此一來, 若 ForkJoinPool 的 thread 只有一個, MainTask 佔一個, 那 subtask 要 fork 再 join 不就沒有 thread 可以處理?

測試之後發現, MainTask 的 thread 其實會在呼叫 join 的時候就被交出去.
因此當只有一個 thread, MainTask 的 thread 會執行到呼叫 subtask.join 的時候就離開, 讓出 thread 去執行 subtask 的任務.
舉例來說
public class ForkJoinSingleThreadMain {

    private static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(1);
        pool.submit(new MainTask());
        latch.await();
    }

    private static class MainTask extends RecursiveAction {

        @Override
        protected void compute() {
            Subtask subtask1 = new Subtask(1);
            Subtask subtask2 = new Subtask(2);
            System.out.println(Thread.currentThread().getName() + " fork subtask1");
            subtask1.fork();
            System.out.println(Thread.currentThread().getName() + " fork subtask2");
            subtask2.fork();
            System.out.println(Thread.currentThread().getName() + " join subtask1");
            subtask1.join();
            System.out.println(Thread.currentThread().getName() + " join subtask2");
            subtask2.join();
            System.out.println(Thread.currentThread().getName() + " join done");
            latch.countDown();
        }
    }

    private static class Subtask extends RecursiveAction {

        private final int id;

        Subtask(int id) {
            this.id = id;
        }

        @Override
        protected void compute() {
            System.out.println(Thread.currentThread().getName() + ":" + id + " compute");
            try {
                TimeUnit.MILLISECONDS.sleep(300);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread().getName() + ":" + id + " compute done");
        }
    }


}
這個 class 的執行結果是:
可以看到當 MainTask 呼叫 join 的時候就把 thread 交出去了.

那會 deadlock 嗎?
其實就是看 subtask 是否要搶 lock, 如果有就可能會 deadlock.
下面這個例子就可以製造 deadlock.
不過 ForkJoinPool 不能宣告為 new ForkJoinPool(1) 因為這樣只會有一個 thread 執行
至少要 new ForkJoinPool(2) 以上, 才可以有兩個 thread 搶 lock 造成 deadlock

public class ForkJoinPoolDeadlockMain {

    private static final CountDownLatch l = new CountDownLatch(1);
    private static final Object lock1 = new Object();
    private static final Object lock2 = new Object();

    public static void main(String[] params) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        pool.submit(ForkJoinTask.adapt(new Runnable(){
            @Override
            public void run() {
                Obj1Locker locker1 = new Obj1Locker();
                Obj2Locker locker2 = new Obj2Locker();
                locker1.fork();
                locker2.fork();
                locker1.join();
                locker2.join();
                l.countDown();
            }
        }));
        l.await();
    }

    private static class Obj1Locker extends RecursiveAction {

        @Override
        public void compute() {
            synchronized (lock1) {
                System.out.println("obj1Locker got lock1");
                sleep();
                synchronized (lock2) {
                    System.out.println("obj1Locker got lock2");
                }
            }
        }


    }

    private static class Obj2Locker extends RecursiveAction {

        @Override
        public void compute() {
            synchronized (lock2) {
                System.out.println("obj2Locker got lock2");
                sleep();
                synchronized (lock1) {
                    System.out.println("obj2Locker got lock1");
                }
            }
        }


    }

    private static void sleep() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
        }
    }

}

Java Concurrent - ForkJoinPool

ExecutorService and Problem

用一個 ExecutorService submit task 後, 如果 task 會 submit 其他的 task, 這些 task 都會放進 ExecutorService 的 queue 中.
這樣就沒辦法做到 "第一個 task 與其產生的 subtask 完成後就印一行 log"

ForkJoinPool and Purpose

使用了 ForkJoinPool 就可以透過 task 的 fork 與 join 讓 task 變成樹狀結構.
  1. 建立 ForkJoinPool
  2. 用 ForkJoinPool submit task
  3. task 建立幾個新的 subtask
  4. 呼叫 subtask 的 fork
  5. 呼叫 subtask 的 join
  6. 執行到 parent 的 task 結束
如此就能讓 task 自然分群

Reference and Sample

Code

public class ForkJoinPoolMain {

    private static AtomicInteger actionCounter = new AtomicInteger();
    public static void main(String[] params) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        System.out.println("main:" + pool.submit(new MyAction(new int[]{1,2,3,4,5,6,7,8,9,10,11})).get());
    }

    private static class MyAction extends RecursiveTask<Integer> {
        // Extends RecursiveAction if no return value required

        private int count;
        private int[] numbers;

        public MyAction(int[] numbers) {
            this.count = actionCounter.incrementAndGet();
            this.numbers = numbers;
        }

        @Override
        protected Integer compute() {
            if (numbers.length > 2) {
                System.out.println(count + " split " + Arrays.toString(numbers) + " to 2 arrays");
                List<MyAction> actions = new ArrayList<>();
                actions.add(new MyAction(Arrays.copyOfRange(numbers, 0, numbers.length / 2)));
                actions.add(new MyAction(Arrays.copyOfRange(numbers, numbers.length / 2, numbers.length)));
                actions.forEach(MyAction::fork);
                return actions.stream().mapToInt(MyAction::join).sum();
            } else if (numbers.length == 2) {
                System.out.println(count + " add " + numbers[0] + " and " + numbers[1]);
                return numbers[0] + numbers[1];
            }
            System.out.println(count + " just return " + numbers[0]);
            return numbers[0];
        }
    }

}
下篇:ForkJoinPool 沒處理好也會 deadlock

Java Concurrent CyclicBarrier

Introduction

CyclicBarrier 很久沒用都忘了, 來複習一下..

Description

簡單說就是很多個 thread 去作事情, 作完之後就 await.
CyclicBarrier 等作完的 thread 達到指定的個數後, await 就結束.

Code

public class CyclicBarrierMain {

    public static void main(String[] params) throws InterruptedException {
        int runner = 5;
        CountDownLatch gameOver = new CountDownLatch(runner);
        CyclicBarrier b = new CyclicBarrier(runner, () -> {
            System.out.println("barrier done");
        });
        IntStream.range(0,runner).forEach(idx -> {
            new Thread() {
                @Override
                public void run() {
                    try {
                        System.out.println("wait " + idx);
                        b.await();
                        System.out.println("count down " + idx);
                        gameOver.countDown();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        });
        gameOver.await();
        System.out.println("game over");
    }

}

Java Concurrent Phaser

Introduction

今天才知道 Java 7 有提供一個 Phaser class 來處理更複雜的 CountDownLatch 需求.

Simple Phaser

想用 Phaser, 首先你的 threads 工作要有分階段, 一個階段沒做完就不能進行下個階段.
有了這個限制後就可以照下面的流程做事情
  1. 先註冊, 看有幾個 thread 要做事情.
    => 透過呼叫 Phaser#register 去註冊
  2. 做完之後呼叫 arrive, 告訴 Phaser 有事情做完了
  3. 當每個 thread 都把事情做完, Phaser 就會進入下一個階段, 稱做 advance
  4. 當進入下一階段, Phaser 的 onAdvance method 就會被呼叫,
    這個 method 是留給我們 developer 去實做
以上. 簡單的概念, 覺得看好久都沒看懂, 希望看這篇的人可以簡單看懂.
接著是範例, 這範例是

  1. 假設有幾個人賽跑
  2. 首先要到達起跑線 
  3. 都到了以後才開跑 
  4. 每個人都到達終點舊比賽結束.

public class PhaserRacingGame {

    public static void main(String[] params) {
        Random r = new Random();
        Phaser perGamePhase = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("phase:" + phase + ", registeredParties:" + registeredParties + ", arrived:" + getArrivedParties());
                switch (phase) {
                    case 0:
                        System.out.println("start!");
                        break;
                    case 1:
                        System.out.println("game finish");
                        break;
                    default:
                        System.out.println("unexpected:" + phase);
                }
                return super.onAdvance(phase, registeredParties);
            }
        };
        IntStream.range(0,5).forEach(idx -> {
            perGamePhase.register();
            new Thread(() -> {
                System.out.println(idx + " arrive starting line and wait for start");
                perGamePhase.arriveAndAwaitAdvance();
                System.out.println(idx + " run!! startTime:" + System.currentTimeMillis());
                long spendTime = r.nextInt(1000);
                sleep(spendTime);
                perGamePhase.arrive();
            }).start();
        });
        sleep(5000);
    }

    private static void sleep(long ms) {
        try {
            TimeUnit.MILLISECONDS.sleep(ms);
        } catch (InterruptedException e) {
        }
    }

}

Parent Phaser

再來比較複雜的是 Parent Phaser, Parent Phaser 的目的是減少 synchronization 的時候搶 lock 的碰撞. 
使用 Parent Phaser 的時候要注意的是: advance 是發生在 Parent Phaser 而不是 Child Phaser, 
使用 Parent Phaser 的時候, 是可以確保每個 Child Phaser 準備號都進入下一個 phase 的時候, 已經先到達的 Child Phaser 才可以更進一步 arrive.
比方說世足十六強賽, 要等所有能進入八強賽的隊伍都出線之後才可以開始進行八強賽.
那 Parent Phaser 就是賽程, 十六強賽則是有八個 Child Phaser.
以下範例是看出特性, 方便 copy paste 觀察
public class ParentPhaser {

    public static void main(String[] params) {
        Phaser parent = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("parent phase:" + phase + ", registeredParties:" + registeredParties);
                return super.onAdvance(phase, registeredParties);
            }
        };
        Phaser child1 = new Phaser(parent, 4) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("child phase:" + phase + ", registeredParties:" + registeredParties);
                return super.onAdvance(phase, registeredParties);
            }
        };
        Phaser child2 = new Phaser(parent, 3) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("child phase:" + phase + ", registeredParties:" + registeredParties);
                return super.onAdvance(phase, registeredParties);
            }
        };
        IntStream.range(0,10).forEach(idx -> {
            child1.arrive();
            child1.arrive();
            child1.arrive();
            child1.arrive();
//            child1.arrive(); // cause error, child1 need wait for child2 finish
            child2.arrive();
            child2.arrive();
            child2.arrive();
        });
    }


}

Zip files

In a project, I need zip file and append to separated folders.

public static void main(String[] params) throws IOException {
 List<String> srcFiles = Arrays.asList("E:\\file1.jar", 
   "E:\\file2.jar");
 FileOutputStream fos = new FileOutputStream("E:\\targetfile.zip");
 ZipOutputStream zipOut = new ZipOutputStream(fos);
 for (String srcFile: srcFiles) {
  File fileToZip = new File(srcFile);
  System.out.println(fileToZip + ", canRead:" + fileToZip.canRead());
  ZipEntry zipEntry = new ZipEntry(String.valueOf(srcFiles.indexOf(srcFile)) + "/" + fileToZip.getName());
  zipOut.putNextEntry(zipEntry);

  FileInputStream fis = new FileInputStream(fileToZip);
  IOUtils.copy(fis, zipOut);
  IOUtils.closeQuietly(fis);
 }
 IOUtils.closeQuietly(zipOut);
 IOUtils.closeQuietly(fos);
}

Scan log and paste to Google sheet to know the Flush pressure

Introduction

After a performance issue happen, we have less information about system status at that time.
There is flush log in Cassandra, so I write code to parse it and write to output file.
Thus I can paste the parse result to google sheet or excel to draw charts such as:

Code

public  class  ScanLog  {

  private  static  final  String  fileSufix  =  "10";
  private  static  final  String  LOG_FOLDER  =  "/cassandraLogPath";
  private  static  final  String  OUTPUT_FOLDER  =  "/outputpath";

  public  static  void  main(String[]  params)  throws  IOException,  ParseException  {
    File  folder  =  new  File(LOG_FOLDER);
    Map<String,List<CompactSize>>  cf2CompactSizes  =  toCompactSizes(folder);
    FileUtils.write(new  File(OUTPUT_FOLDER  +  "output-compact-mb-"  +  fileSufix  +  ".csv"),  toCompactSizesString(cf2CompactSizes),  "UTF-8");
    List<CompactSize>  aggCompactSizes  =  aggregate(toCompactSizeList(folder));
    FileUtils.write(new  File(OUTPUT_FOLDER  +  "output-compact-mb-agg-"  +  fileSufix  +  ".csv"),  toCompactSizesString(aggCompactSizes),  "UTF-8");
    List<FlushEvent>  events  =  toEvents(folder);
    FileUtils.write(new  File(OUTPUT_FOLDER  +  "output-queue-"  +  fileSufix  +  ".csv"),  toString(events),  "UTF-8");
    Map<String,  List<Enqueue>>  cfToE  =  cfToE(folder);
    Map<String,  LinkedHashMap<String,Long>>  cfToTimeToAvg  =  cfToTimeToAvg(cfToE);
    FileUtils.write(new  File(OUTPUT_FOLDER  +  "output-"  +  fileSufix  +  ".csv"),  toString(cfToE),  "UTF-8");
    FileUtils.write(new  File(OUTPUT_FOLDER  +  "output-"  +  fileSufix  +  "-avg.csv"),  toAvgString(cfToTimeToAvg),  "UTF-8");
  }

  private  static  List<CompactSize>  aggregate(List<CompactSize>  compactSizes)  {
    LinkedHashMap<String,List<CompactSize>>  minutes2CompactSizes  =  new  LinkedHashMap<>();
    compactSizes.forEach(compactSize  ->  {
      String  timeMinutes  =  compactSize.getTimeMinutes();
      if  (!minutes2CompactSizes.containsKey(timeMinutes))  {
        minutes2CompactSizes.put(timeMinutes,  new  ArrayList<>());
      }
      minutes2CompactSizes.get(timeMinutes).add(compactSize);
    });  
    LinkedHashMap<String,CompactSize>  minutes2AggregatedSize  =  new  LinkedHashMap<>();
    minutes2CompactSizes.entrySet().forEach(entry  ->  {
      String  minutes  =  entry.getKey();
      entry.getValue().forEach(size  ->  {
        if  (!minutes2AggregatedSize.containsKey(size.getTimeMinutes()))  {
          minutes2AggregatedSize.put(size.getTimeMinutes(),new  CompactSize(size.getTime(),size.getCf(),size.getDataSize()));
        }  else  {
          CompactSize  total  =  minutes2AggregatedSize.get(size.getTimeMinutes());
          CompactSize  newTotal  =  new  CompactSize(size.getTime(),size.getCf(),  size.getDataSize()  +  total.getDataSize());
          minutes2AggregatedSize.put(size.getTimeMinutes(),  newTotal);
        }
      });
    });
    List<CompactSize>  result  =  new  ArrayList<>();
    minutes2AggregatedSize.entrySet().forEach(entry  ->  {
      result.add(entry.getValue());
    });
    Collections.sort(result,  Comparator.comparingLong(CompactSize::getTimeMillis));
    return  result;
  }
  
  private  static  String  toCompactSizesString(List<CompactSize>  cf2compactSizes)  {
    StringBuilder  sb  =  new  StringBuilder();
    cf2compactSizes.forEach(size  ->  {
      sb.append(size.getCf()  +  ","  +  size.getTime()  +  ","  +  size.getDataSizeInMB()).append("\n");
    });
    return  sb.toString();
  }
  
  private  static  String  toCompactSizesString(Map<String,List<CompactSize>>  cf2compactSizes)  {
    StringBuilder  sb  =  new  StringBuilder();
    cf2compactSizes.keySet().forEach(cf  ->  {
      cf2compactSizes.get(cf).forEach(size  ->  {
        sb.append(size.getCf()  +  ","  +  size.getTime()  +  ","  +  size.getDataSizeInMB()).append("\n");
      });
    });
    return  sb.toString();
  }

  private  static  List<CompactSize>  toCompactSizeList(File  folder)  {
    List<CompactSize>  result  =  new  ArrayList<>();
    Arrays.stream(folder.listFiles())
        .filter(file  ->  file.getName().contains("system.log"))
        .forEach(file  ->  {
          try  {
            for  (String  line:  FileUtils.readLines(file))  {
              if  (  CompactSize.isMyPattern(line)  )  {
                CompactSize  compactSize  =  new  CompactSize(line);
                result.add(compactSize);
              }
            }
          }  catch  (Exception  ex)  {
            ex.printStackTrace();
          }
        });
    Collections.sort(result,  Comparator.comparingLong(CompactSize::getTimeMillis));
    return  result;
  }
  
  private  static  Map<String,List<CompactSize>>  toCompactSizes(File  folder)  {
    Map<String,List<CompactSize>>  cf2CompactSizes  =  new  HashMap<>();
    Arrays.stream(folder.listFiles())
        .filter(file  ->  file.getName().contains("system.log"))
        .forEach(file  ->  {
          try  {
            for  (String  line:  FileUtils.readLines(file))  {
              if  (  CompactSize.isMyPattern(line)  )  {
                CompactSize  compactSize  =  new  CompactSize(line);
                if  (!cf2CompactSizes.containsKey(compactSize.getCf()))  {
                  cf2CompactSizes.put(compactSize.getCf(),  new  ArrayList<>());
                }
                cf2CompactSizes.get(compactSize.getCf()).add(compactSize);
              }
            }
          }  catch  (Exception  ex)  {
            ex.printStackTrace();
          }
        });
    cf2CompactSizes.values().forEach(compactSizes  ->  
        Collections.sort(compactSizes,  Comparator.comparingLong(CompactSize::getTimeMillis)));
    return  cf2CompactSizes;
  }
  
  private  static  class  CompactSize  {
    private  final  String  time;
    private  final  String  cf;
    private  final  long  dataSize;
    
    CompactSize(String  time,  String  cf,  long  dataSize)  {
      this.time  =  time;
      this.cf  =  cf;
      this.dataSize  =  dataSize;
    }
    
    CompactSize(String  line)  {
      this.time  =  substringBetween(line,  "]  ",",");
      this.cf  =  substringBetween(line,  "/data/ng/db/data/wsg/",  "/");
      this.dataSize  =  NumberUtils.toInt(StringUtils.replace(substringBetween(line,  "].    ",  "  bytes"),",",""));
    }

    public  static  boolean  isMyPattern(String  line)  {
      return  StringUtils.contains(line,  "CompactionTask.java  (line  302)  Compacted")
          &&  StringUtils.contains(line,  "[/data/ng/db/data/wsg/");
    }
    
    public  long  getDataSizeInMB()  {
      return  dataSize  //  bytes  
          /  1024    //  KB
          /  1024;  //  MB
    }
    
    public  long  getDataSize()  {
      return  dataSize;
    }

    public  String  getCf()  {
      return  cf;
    }

    public  String  getTimeMinutes()  {
      long  millis  =  getTimeMillis();
      return  sdfToMinutes.format(new  Date(millis));
    }
    
    public  Long  getTimeMillis()  {
      return  toMillis(getTime());
    }
    
    public  String  getTime()  {
      return  time;
    }
  }
  
  private  static  String  toString(List<FlushEvent>  events)  {
    StringBuilder  sb  =  new  StringBuilder();
    AtomicInteger  waitForWrite  =  new  AtomicInteger();
    AtomicInteger  writing  =  new  AtomicInteger();
    events.forEach(e  ->  {
      switch  (  e.getType()  )  {
        case  Enqueue:
          waitForWrite.incrementAndGet();
          break;
        case  Write:
          writing.incrementAndGet();
          waitForWrite.decrementAndGet();
          break;
        case  Complete:
          writing.decrementAndGet();
          break;
        default:  
          throw  new  RuntimeException("Shouldn't  happen");
      }
      sb.append(e.getTime()  +  ","  +  waitForWrite  +  ","  +  writing).append("\n");
    });
    return  sb.toString();
  }
  
  private  static  List<FlushEvent>  toEvents(File  folder)  {
    List<FlushEvent>  events  =  new  ArrayList<>();
    Arrays.stream(folder.listFiles())
        .filter(file  ->  file.getName().contains("system.log"))
        .forEach(file  ->  {
          try  {
            for  (String  line:  FileUtils.readLines(file))  {
              Optional<FlushEventType>  type  =  FlushEventType.getByPattern(line);
              type.ifPresent(t  ->  {
                events.add(new  FlushEvent(line));
              });
            }
          }  catch  (Exception  ex)  {
            ex.printStackTrace();
          }
          });
    Collections.sort(events,  Comparator.comparingLong(FlushEvent::getTimeMillis));
    return  events;
  }

  private  static  String  toAvgString(Map<String,  LinkedHashMap<String,Long>>  cfToTimeToAvg)  {
    StringBuilder  sb  =  new  StringBuilder();
    cfToTimeToAvg.keySet().forEach(cf  ->  {
      cfToTimeToAvg.get(cf).forEach((time,avg)  ->  {
        sb.append(cf  +  ","  +  time  +  ","  +  avg).append("\n");
      });
    });
    return  sb.toString();
  }

  private  static  String  toString(Map<String,  List<Enqueue>>  cfToE)  {
    StringBuilder  sb  =  new  StringBuilder();
    cfToE.values().stream().forEach(eList  ->  {
      eList.stream().forEach(e  ->  {
        try  {
          sb.append(e.getCf()  +  ","  +  e.getTime()  +  ","  +  e.serializedMB).append("\n");
        }  catch  (Exception  ex)  {
          ex.printStackTrace();
        }
      });
    });
    return  sb.toString();
  }

  private  static  Map<String,  LinkedHashMap<String,Long>>  cfToTimeToAvg(Map<String,  List<Enqueue>>  cfToE)  {
    Map<String,  LinkedHashMap<String,Long>>  cfToTimeToAvg  =  new  HashMap<>();
    cfToE.keySet().forEach(cf  ->  {
      if  (!cfToTimeToAvg.containsKey(cf))  {
        cfToTimeToAvg.put(cf,  new  LinkedHashMap<>());
      }
      LinkedHashMap<String,List<Long>>  timeToMBList  =  new  LinkedHashMap<>();
      cfToE.get(cf).forEach(enqueue  ->  {
        if  (!timeToMBList.containsKey(enqueue.getTimeMinutes()))  {
          timeToMBList.put(enqueue.getTimeMinutes(),  new  ArrayList<>());
        }
        timeToMBList.get(enqueue.getTimeMinutes()).add(enqueue.serializedMB);
      });
      AtomicReference<String>  previousTime  =  new  AtomicReference<>();
      LinkedHashMap<String,Long>  timeToAvg  =  cfToTimeToAvg.get(cf);
      timeToMBList.keySet().forEach(time  ->  {
        try  {
          if  (previousTime.get()  !=  null)  {
            AtomicLong  totalMB  =  new  AtomicLong();
            timeToMBList.get(time).forEach(mb  ->  {
              totalMB.addAndGet(mb);
            });
            long  currentTimeMillis  =  sdfToMinutes.parse(time).getTime();
            long  previousTimeMillis  =  sdfToMinutes.parse(previousTime.get()).getTime();
            long  gapMinutes  =  TimeUnit.MILLISECONDS.toMinutes(currentTimeMillis  -  previousTimeMillis);
            timeToAvg.put(time,  (totalMB.longValue()  /  gapMinutes));
          }
          previousTime.set(time);
        }  catch  (Exception  ex)  {
          ex.printStackTrace();
        }
      });
    });
    return  cfToTimeToAvg;
  }

  private  static  Map<String,  List<Enqueue>>  cfToE(File  folder)  {
    Map<String,List<Enqueue>>  cfToE  =  new  HashMap<>();
    Arrays.stream(folder.listFiles())
        .filter(f  ->  f.getName().contains("system.log"))
        .forEach(file  ->  {
          try  {
            List<String>  lines  =  IOUtils.readLines(new  FileInputStream(file));
            for  (String  line:  lines)  {
              if  (StringUtils.contains(line,  "Enqueuing  flush  of  Memtable-"))  {
                Enqueue  e  =  new  Enqueue(line);
                if  (e.serializedMB  <=  1)  {
                  continue;
                }
                if  (!cfToE.containsKey(e.getCf()))  {
                  cfToE.put(e.getCf(),  new  ArrayList<>());
                }
                cfToE.get(e.getCf()).add(e);
              }
            }
          }  catch  (Exception  ex)  {
            ex.printStackTrace();
          }
        });
    cfToE.keySet().forEach(k  ->  {
      cfToE.get(k).sort((e1,e2)  ->  {
        try  {
          return  (int)(e1.getTimeMillis()-e2.getTimeMillis());
        }  catch  (Exception  ex)  {
          ex.printStackTrace();
          return  0;
        }
      });
    });
    return  cfToE;
  }

  private  enum  FlushEventType  {
    Enqueue("Enqueuing  flush  of  Memtable-"),
    Write("Writing  Memtable-"),
    Complete("Completed  flushing  ");
    private  final  String  pattern;
    FlushEventType(String  pattern)  {
      this.pattern  =  pattern;
    }

    public  String  getPattern()  {
      return  pattern;
    }

    public  static  Optional<FlushEventType>  getByPattern(String  line)  {
      return  Arrays.stream(values())
          .filter(type  ->  StringUtils.contains(line,  type.getPattern()))
          .findFirst();
    }
  }
  
  private  static  class  FlushEvent  {
    private  final  FlushEventType  type;
    private  final  String  time;
    FlushEvent(String  line)  {
      this.time  =  substringBetween(line,  "]  ",",");
      this.type  =  FlushEventType.getByPattern(line).get();  //  must  success
    }

    public  FlushEventType  getType()  {
      return  type;
    }

    public  Long  getTimeMillis()  {
      return  toMillis(getTime());
    }
    
    public  String  getTime()  {
      return  time;
    }
  }
  
  private  static  class  Enqueue  {
    private  final  String  time;
    private  final  String  cf;
    private  final  int  ops;
    private  final  long  serializedMB;

    Enqueue(String  line)  throws  ParseException  {
      this.time  =  substringBetween(line,  "]  ",",");
      this.cf  =  substringBetween(line,  "Enqueuing  flush  of  Memtable-",  "@");
      this.ops  =  NumberUtils.toInt(substringBetween(line,  "serialized/live  bytes,  ",  "  ops)"));
      this.serializedMB  =  NumberUtils.toInt(substringBetween(substringAfter(line,  "Enqueuing  flush  of  Memtable-"),  "(",  "/"))  /  1024  /  1024;
    }

    @Override
    public  String  toString()  {
      return  new  ToStringBuilder(this)
          .append("time",time)
          .append("cf",cf)
          .append("ops",ops)
          .append("serializedMB",  serializedMB)
          .toString();
    }

    public  String  getTimeMinutes()  {
      long  millis  =  getTimeMillis();
      return  sdfToMinutes.format(new  Date(millis));
    }

    public  Long  getTimeMillis()  {
      return  toMillis(getTime());
    }

    public  String  getTime()  {
      return  time;
    }

    public  String  getCf()  {
      return  cf;
    }

  }

  private  static  final  SimpleDateFormat  sdfToMinutes  =  new  SimpleDateFormat("yyyy-MM-dd  HH:mm");
  private  static  final  SimpleDateFormat  sdf  =  new  SimpleDateFormat("yyyy-MM-dd  HH:mm:SS");

  private  static  long  toMillis(String  time)  {
    try  {
      Date  date  =  sdf.parse(time);
      return  date.getTime();
    }  catch  (Exception  ex)  {
      throw  new  RuntimeException(ex);
    }
  }

}

lombok + Orika + Builder Pattern

Reference


Introduction

  1. Builder Pattern is suggested in Effective Java and used to generate immutable object.
  2. Builder Pattern introduces many codes, lombok project makes it easier
  3. We often use Orika to map data when transfer object from UI entity to domain entity or data persistence object to domain entity.
  4. It's hard to map Java Bean to Builder Pattern object because Builder Pattern doesn't follows Java Bean convention. It causes we can't leverage lombok and Orika at the same time

How to fix

  1. Copy and paste the following code: BuilderPropertyResolver
  2. /**
     * This class is used to map from an object to a builder who follows Builder Pattern.
     * 
     * <pre>
     * MapperFactory factory = new DefaultMapperFactory.Builder()
     *  .propertyResolverStrategy(new BuilderPropertyResolver())
     *  .build();
     *  factory.registerObjectFactory((Object o, MappingContext mappingContext) 
     *  -&gt; new MyBuilder(), TypeFactory.valueOf(MyBuilder.class));
     *  </pre>
     */
    public class BuilderPropertyResolver extends IntrospectorPropertyResolver {
    
     private static final Logger logger = LoggerFactory.getLogger(BuilderPropertyResolver.class);
     
     @Override
     protected void collectProperties(Class<?> type, Type<?> referenceType, Map<String, Property> properties) {
      super.collectProperties(type, referenceType, properties);
      if (StringUtils.endsWith(type.getName(), "Builder")) {
       Set<String> fieldNames = Arrays.stream(type.getDeclaredFields()).map(Field::getName).collect(Collectors.toSet());
       Arrays.stream(type.getDeclaredMethods())
         .filter(method -> fieldNames.contains(method.getName()))
         .filter(method -> isWriteMethodInBuilder(type, method))
         .forEach(method -> {
          Property.Builder builder = new Property.Builder();
          builder.expression(method.getName());
          builder.name(method.getName());
          builder.setter(method.getName() + "(%s)");
          Class<?> fieldType = method.getParameterTypes()[0];
          builder.type(this.resolvePropertyType(null, fieldType, type, referenceType));
          Property property = builder.build(this);
          properties.put(method.getName(), property);
         });
      }
     }
    
     private boolean isWriteMethodInBuilder(Class<?> type, Method method) {
      try {
       Class<?> returnType = method.getReturnType();
       Class<?> fieldType = type.getDeclaredField(method.getName()).getType();
       boolean onlyOneParameter = method.getParameterTypes().length == 1;
       boolean methodReturnBuilderType = returnType.equals(type);
       Class<?> methodParamType = method.getParameterTypes()[0];
       boolean isFieldSetter = fieldType.equals(methodParamType);
       return methodReturnBuilderType && onlyOneParameter && isFieldSetter;
      } catch (NoSuchFieldException e) {
       e.printStackTrace();
       return false;
      }
     }
    
    }
    
  3. Leverage this object
    public class Mapper {
     
     public static void main(String[] params) {
      MapperFactory factory = new DefaultMapperFactory.Builder()
        .propertyResolverStrategy(new BuilderPropertyResolver())
        .build();
      factory.registerObjectFactory((Object o, MappingContext mappingContext) 
        -> Person.builder(), TypeFactory.valueOf(Person.PersonBuilder.class));
      
      
      PersonBO bo = new PersonBO();
      bo.setName("TEST");
      factory.classMap(PersonBO.class, Person.class);
      System.out.println(factory.getMapperFacade().map(bo, Person.PersonBuilder.class).build());
      
      Person.PersonBuilder builder = Person.builder().list(Arrays.asList("A"));
      factory.getMapperFacade().map(bo, builder);
      System.out.println(builder);
     }
    
     @Data
     public static class PersonBO {
      private String name;
      private List<String> list;
     }
     
     @ToString
     @Builder
     public static class Person {
      private String name;
      private List<String> list;
     }
     
    }
    

Handle InterruptedException

Reference


How to handle InterruptedException

  • Throw it directly
  • If you can't throw it, call Thread.currentThread().interrupt()
  • Check Thread.currentThread().isInterrupted() every while loop if you catch an InterruptedException within the loop block

Example:

This example shows how to handle InterruptedException, you can change code to check what will happen in other cases.
public class TestMain {

 public static void main(String[] params) {
  ThreadPoolExecutor e = (ThreadPoolExecutor) Executors.newCachedThreadPool();
  MyRunnable r = new MyRunnable();
  e.execute(r);
  System.out.println("shutdown");
  e.shutdownNow();
  System.out.println("shutdown done");
 }

 private static class MyRunnable implements Runnable {

  @Override
  public void run() {
   while (!Thread.currentThread().isInterrupted()) {
    try {
     System.out.println("sleep");
     TimeUnit.HOURS.sleep(1);
    } catch (InterruptedException e) {
     e.printStackTrace();
     Thread.currentThread().interrupt();
    }
   }
  }
 }

}

Java TLS Practice

Introduction

記錄 Java 使用 SSL 的方法 (不要求 client auth)

Reference

Prepare jks files

  1. Install openssl in Ubuntu
    sudo apt-get install openssl
  2. Execute following commands one by one
    openssl req -x509 -newkey rsa:1024 -keyout private/cakey.pem -out private/cacert.pem -days 3650
    openssl x509 -in private/cacert.pem -addtrust clientAuth -setalias "Isaac Test Class 1 CA" -out public/catrust.pem
    keytool -importcert -trustcacerts -noprompt -file private/cacert.pem -alias ca -keystore public/catrust.jks -storepass 123456
    keytool -genkeypair -keyalg RSA -keysize 1024 -validity 730 -keystore public/server.jks
    keytool -certreq -file server-req.pem -keystore public/server.jks
    openssl x509 -req -in server-req.pem -out public/server-cert.pem -CA private/cacert.pem -CAkey private/cakey.pem -extensions v3_usr -days 730 -CAserial private/cacert.srl -CAcreateserial
    cat private/cacert.pem >> public/server-cert.pem
    keytool -importcert -v -file public/server-cert.pem -keystore public/server.jks
    rm server-req.pem
  3. There are public/server.jks and public/catrust.jks

Run server code

public class ServerDontNeedClientAuth {

    private static String SERVER_KEY_STORE = Paths.get("src/main/resources/certs/server.jks").toAbsolutePath().toString();
    private static String SERVER_KEY_STORE_PASSWORD = "123456";

    public static void main(String[] params) throws Exception {
//        System.setProperty("javax.net.debug", "ssl,handshake");
        System.setProperty("javax.net.ssl.trustStore", SERVER_KEY_STORE);
        SSLContext context = SSLContext.getInstance("TLS");
        KeyStore ks = KeyStore.getInstance("jceks");
        ks.load(new FileInputStream(SERVER_KEY_STORE), null);
        KeyManagerFactory kf = KeyManagerFactory.getInstance("SunX509");
        kf.init(ks, SERVER_KEY_STORE_PASSWORD.toCharArray());
        context.init(kf.getKeyManagers(), null, null);

        ServerSocketFactory factory = context.getServerSocketFactory();
        ServerSocket serverSocket = factory.createServerSocket(8443);
        SSLServerSocket sslServerSocket =  (SSLServerSocket) serverSocket;
        sslServerSocket.setNeedClientAuth(false);
        while(true){
            try{
                System.out.println("listen port 8443..");
                Socket socket = sslServerSocket.accept();
                System.out.println("accept:" + socket);
                InputStream is = socket.getInputStream();
                BufferedReader buffer = new BufferedReader(new InputStreamReader(is));
                String readLine = buffer.readLine();
                System.out.println("server receive:" + readLine);
                socket.getOutputStream().write("1234567890".getBytes());
                socket.close();
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }

}

Run client code

public class ClientWithoutAuth {

    private static String CLIENT_KEY_STORE = Paths.get("src/main/resources/certs/catrust.jks").toAbsolutePath().toString();

    public static void main(String[] params) throws Exception {
        System.setProperty("javax.net.ssl.trustStore", CLIENT_KEY_STORE);
        SocketFactory sf = SSLSocketFactory.getDefault();
        Socket s = sf.createSocket("localhost", 8443);
        PrintWriter writer = new PrintWriter(s.getOutputStream());
        BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
        writer.println("hello\n");
        writer.flush();
        System.out.println("client receive:" + reader.readLine());
        s.close();
    }

}

Server output

listen port 8443..
accept:340f438e[SSL_NULL_WITH_NULL_NULL: Socket[addr=/127.0.0.1,port=50429,localport=8443]]
server receive:hello
listen port 8443..

Client output

1234567890

guava notes

Reference

Optional

Optional 可以用來處理 null value 與避免 NullPointerException
public class OptionalTest {


    @Test
    public void test() {
        try {
            Optional.of(null);
            Assert.fail("Optional.of shouldn't accept null");
        } catch (NullPointerException ex) {
        }
        Optional opt = Optional.of("Test");
        Assert.assertTrue(opt.isPresent());
        Assert.assertEquals("Test", opt.get());
        Assert.assertEquals("Test", opt.or("QQ"));
        Assert.assertEquals("Test", opt.orNull());
        Assert.assertEquals(ImmutableSet.of("Test"), opt.asSet());

        Optional nullableOptional = Optional.fromNullable(null);
        Assert.assertFalse(nullableOptional.isPresent());
        try {
            Assert.assertNull(nullableOptional.get());
            Assert.fail();
        } catch (IllegalStateException ex) {
        }
        Assert.assertEquals("QQ", nullableOptional.or("QQ"));
        Assert.assertNull(nullableOptional.orNull());
        Assert.assertEquals(ImmutableSet.of(), nullableOptional.asSet());
        Assert.assertEquals(Optional.absent(), nullableOptional);
    }

}

ComparisonChain

ComparisonChain 可以用比較漂亮的程式實作 compare
public class ComparisonChainTest {

    @Test
    public void test() {
        NumbersObject n1 = new NumbersObject(1,2,3);
        NumbersObject n2 = new NumbersObject(1,2,4);
        NumbersObject n3 = new NumbersObject(1,3,4);
        NumbersObject n4 = new NumbersObject(2,2,4);

        List<NumbersObject> list = Arrays.asList(n4, n3, n2, n1);
        Collections.sort(list, (o1, o2) -> ComparisonChain.start()
                .compare(o1.a,o2.a)
                .compare(o1.b,o2.b)
                .compare(o1.c,o2.c)
                .result());
        Assert.assertTrue(n1 == list.get(0));
        Assert.assertTrue(n2 == list.get(1));
        Assert.assertTrue(n3 == list.get(2));
        Assert.assertTrue(n4 == list.get(3));
    }

    final static class NumbersObject {
        private final int a ;
        private final int b;
        private final int c;
        NumbersObject(int a, int b, int c) {
            this.a = a;
            this.b = b;
            this.c = c;
        }
    }

}

Ordering

Ordering 其實就是 Comparator, 可以傳給 Collections. 例如: Collections.sort(list, Ordering)

usingToString

用 toString 的結果排序
@Test
public void usingToString() {
    NumbersObject n1 = new NumbersObject(1, 2, 3);
    NumbersObject n2 = new NumbersObject(1, 2, 4);
    NumbersObject n3 = new NumbersObject(-1, 2, -3);
    NumbersObject n4 = new NumbersObject(-2, 3, 5);
    List<NumbersObject> numbers = Arrays.asList(n1, n2, n3, n4);
    Collections.sort(numbers, Ordering.usingToString());
    Assert.assertEquals(n3, numbers.get(0));
    Assert.assertEquals(n4, numbers.get(1));
    Assert.assertEquals(n1, numbers.get(2));
    Assert.assertEquals(n2, numbers.get(3));
}

private final class NumbersObject {

    public final Integer a;
    public final Integer b;
    public final Integer c;

    public NumbersObject(Integer a, Integer b, Integer c) {
        this.a = a;
        this.b = b;
        this.c = c;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("a",a)
                .add("b",b)
                .add("c",c)
                .toString();
    }
}

nullsFirst & nullsLast

排序的時候, null 放在最前面位置(或最後).
物件本身是 Comparable, 可以直接拿 nullsFirst 回傳的 Ordering 來排序, 若物件本身不是 Comparable 就要提供排序的方式
@Test
public void nullsFirst_NotComparable() {
    NumbersObject n1 = new NumbersObject(1, 2, 3);
    NumbersObject n2 = new NumbersObject(1, 2, 4);
    NumbersObject n3 = new NumbersObject(-1, 2, -3);
    NumbersObject n4 = new NumbersObject(-2, 3, 5);
    NumbersObject nullNumber = new NumbersObject(null,null,null);
    List<NumbersObject> numbers = Arrays.asList(n1, n2, n3, n4, nullNumber);
    Ordering ordering = Ordering.natural().nullsFirst().onResultOf(new Function<NumbersObject, Integer>() {
        @Override
        public Integer apply(NumbersObject numbersObject) {
            return numbersObject.a;
        }
    });
    Collections.sort(numbers, ordering);
    Assert.assertEquals(nullNumber, numbers.get(0));
    Assert.assertEquals(n4, numbers.get(1));
    Assert.assertEquals(n3, numbers.get(2));
    Assert.assertEquals(n1, numbers.get(3));
    Assert.assertEquals(n2, numbers.get(4));
}

@Test
public void nullsFirst_Comparable() {
    List<Integer> list = Arrays.asList(5, 2, 6, 7, 3, null);
    Collections.sort(list, Ordering.natural().nullsFirst());
    Assert.assertEquals((Integer) null, list.get(0));
    Assert.assertEquals((Integer) 2, list.get(1));
    Assert.assertEquals((Integer) 3, list.get(2));
    Assert.assertEquals((Integer) 5, list.get(3));
    Assert.assertEquals((Integer) 6, list.get(4));
    Assert.assertEquals((Integer) 7, list.get(5));
}

@Test
public void nullsLast_Comparable() {
    List<Integer> list = Arrays.asList(5, 2, 6, 7, 3, null);
    Collections.sort(list, Ordering.natural().nullsLast());
    Assert.assertEquals((Integer) 2, list.get(0));
    Assert.assertEquals((Integer) 3, list.get(1));
    Assert.assertEquals((Integer) 5, list.get(2));
    Assert.assertEquals((Integer) 6, list.get(3));
    Assert.assertEquals((Integer) 7, list.get(4));
    Assert.assertEquals((Integer) null, list.get(5));
}

@Test
public void nullsLast_NotComparable() {
    NumbersObject n1 = new NumbersObject(1, 2, 3);
    NumbersObject n2 = new NumbersObject(1, 2, 4);
    NumbersObject n3 = new NumbersObject(-1, 2, -3);
    NumbersObject n4 = new NumbersObject(-2, 3, 5);
    NumbersObject nullNumber = new NumbersObject(null,null,null);
    List<NumbersObject> numbers = Arrays.asList(n1, n2, n3, n4, nullNumber);
    Ordering ordering = Ordering.natural().nullsLast().onResultOf(new Function<NumbersObject, Integer>() {
        @Override
        public Integer apply(NumbersObject numbersObject) {
            return numbersObject.a;
        }
    });
    Collections.sort(numbers, ordering);
    Assert.assertEquals(n4, numbers.get(0));
    Assert.assertEquals(n3, numbers.get(1));
    Assert.assertEquals(n1, numbers.get(2));
    Assert.assertEquals(n2, numbers.get(3));
    Assert.assertEquals(nullNumber, numbers.get(4));
}

private final class NumbersObject {

    public final Integer a;
    public final Integer b;
    public final Integer c;

    public NumbersObject(Integer a, Integer b, Integer c) {
        this.a = a;
        this.b = b;
        this.c = c;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("a",a)
                .add("b",b)
                .add("c",c)
                .toString();
    }
}

compound

可以組合多個 Comparator, 在上一個 Comparator 無法決定時就交給下一個 Comparator 判斷.
在排列的屬性有優先順序的時候可以讓程式變簡單.
例如: 先排 a, 再 b, 再 c. null 的話就排前面
@Test
public void compound() {
    NumbersObject n1 = new NumbersObject(1, 2, 3);
    NumbersObject n2 = new NumbersObject(1, 2, 4);
    NumbersObject n3 = new NumbersObject(-1, 2, -3);
    NumbersObject n4 = new NumbersObject(-2, 3, 5);
    NumbersObject nullNumber = new NumbersObject(null,null,null);
    List<NumbersObject> numbers = Arrays.asList(n1, n2, n3, n4, nullNumber);
    Comparator<NumbersObject> byNullFirstA = Ordering.natural().nullsFirst().onResultOf((t) -> t.a);
    Comparator<NumbersObject> byNullFirstB = Ordering.natural().nullsFirst().onResultOf((t) -> t.b);
    Comparator<NumbersObject> byNullFirstC = Ordering.natural().nullsFirst().onResultOf((t) -> t.c);
    Ordering orderByABC = Ordering.compound(Arrays.asList(byNullFirstA, byNullFirstB, byNullFirstC));
    Collections.sort(numbers, orderByABC);
    Assert.assertEquals(nullNumber, numbers.get(0));
    Assert.assertEquals(n4, numbers.get(1));
    Assert.assertEquals(n3, numbers.get(2));
    Assert.assertEquals(n1, numbers.get(3));
    Assert.assertEquals(n2, numbers.get(4));
}

private final class NumbersObject {

    public final Integer a;
    public final Integer b;
    public final Integer c;

    public NumbersObject(Integer a, Integer b, Integer c) {
        this.a = a;
        this.b = b;
        this.c = c;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("a",a)
                .add("b",b)
                .add("c",c)
                .toString();
    }
}

lexicographical

跟 Collections.sort 不同, loxicographical 是用來替很多個 iterable 作排序
@Test
public void lexicographically() {
    ImmutableList<String> empty = ImmutableList.of();
    ImmutableList<String> a = ImmutableList.of("a");
    ImmutableList<String> aa = ImmutableList.of("a", "a");
    ImmutableList<String> ab = ImmutableList.of("a", "b");
    ImmutableList<String> b = ImmutableList.of("b");
    List<ImmutableList<String>> all = Arrays.asList(b,a,ab,aa,empty);
    Ordering<Iterable<String>> c = Ordering.<String>natural().lexicographical();
    Collections.sort(all, c);
    Assert.assertEquals(empty, all.get(0));
    Assert.assertEquals(a, all.get(1));
    Assert.assertEquals(aa, all.get(2));
    Assert.assertEquals(ab, all.get(3));
    Assert.assertEquals(b, all.get(4));
}

@Test
public void lexicographicallyReverse() {
    ImmutableList<String> empty = ImmutableList.of();
    ImmutableList<String> a = ImmutableList.of("a");
    ImmutableList<String> aa = ImmutableList.of("a", "a");
    ImmutableList<String> ab = ImmutableList.of("a", "b");
    ImmutableList<String> b = ImmutableList.of("b");
    List<ImmutableList<String>> all = Arrays.asList(b,a,ab,aa,empty);
    Ordering<Iterable<String>> c = Ordering.<String>natural().lexicographical().reverse();
    Collections.sort(all, c);
    Assert.assertEquals(empty, all.get(4));
    Assert.assertEquals(a, all.get(3));
    Assert.assertEquals(aa, all.get(2));
    Assert.assertEquals(ab, all.get(1));
    Assert.assertEquals(b, all.get(0));
}

@Test
public void reverseLexicographically() {
    ImmutableList<String> empty = ImmutableList.of();
    ImmutableList<String> a = ImmutableList.of("a");
    ImmutableList<String> aa = ImmutableList.of("a", "a");
    ImmutableList<String> ab = ImmutableList.of("a", "b");
    ImmutableList<String> b = ImmutableList.of("b");
    List<ImmutableList<String>> all = Arrays.asList(b,a,ab,aa,empty);
    Ordering<Iterable<String>> c = Ordering.<String>natural().reverse().lexicographical();
    Collections.sort(all, c);
    System.out.println(all);
    Assert.assertEquals(empty, all.get(0));
    Assert.assertEquals(b, all.get(1));
    Assert.assertEquals(a, all.get(2));
    Assert.assertEquals(ab, all.get(3));
    Assert.assertEquals(aa, all.get(4));
}

onResultOf

就是自訂排序方式, 直行完後會再往前呼叫其他排序法
@Test
public void onResultOf() {
    Comparator<Integer> c = Ordering.<Integer>natural().onResultOf((x) -> Optional.ofNullable(x).orElse(4)%5);
    List<Integer> list = Arrays.asList(5, 11, 4, 1, 2, 5, 7, 9, null);
    Collections.sort(list, c);
    System.out.println(list);
    Assert.assertEquals(5, list.get(0).intValue());
    Assert.assertEquals(5, list.get(1).intValue());
    Assert.assertEquals(11, list.get(2).intValue());
    Assert.assertEquals(1, list.get(3).intValue());
    Assert.assertEquals(2, list.get(4).intValue());
    Assert.assertEquals(7, list.get(5).intValue());
    Assert.assertEquals(4, list.get(6).intValue());
    Assert.assertEquals(9, list.get(7).intValue());
    Assert.assertEquals(null, list.get(8));
}

@Test
public void onResultOf_reverse() {
    Comparator<Integer> c = Ordering.<Integer>natural().reverse().onResultOf((x) -> Optional.ofNullable(x).orElse(4)%5);
    List<Integer> list = Arrays.asList(5, 11, 4, 1, 2, 5, 7, 9, null);
    Collections.sort(list, c);
    System.out.println(list);
    Assert.assertEquals(5, list.get(8).intValue());
    Assert.assertEquals(5, list.get(7).intValue());
    Assert.assertEquals(11, list.get(5).intValue());
    Assert.assertEquals(1, list.get(6).intValue());
    Assert.assertEquals(2, list.get(3).intValue());
    Assert.assertEquals(7, list.get(4).intValue());
    Assert.assertEquals(4, list.get(0).intValue());
    Assert.assertEquals(9, list.get(1).intValue());
    Assert.assertEquals(null, list.get(2));
}

greatestOf 與 leastOf

拿排序過資料的最後幾筆與最前幾筆.
@Test
public void greatestOf() {
    NumbersObject n1 = new NumbersObject(1, 2, 3);
    NumbersObject n2 = new NumbersObject(1, 2, 4);
    NumbersObject n3 = new NumbersObject(-1, 2, -3);
    NumbersObject n4 = new NumbersObject(-2, 3, 5);
    NumbersObject nullNumber = new NumbersObject(null,null,null);
    List<NumbersObject> numbers = Arrays.asList(n1, n2, n3, n4, nullNumber);
    List<NumbersObject> greatestOf = Ordering.natural().greatestOf(numbers,3);
    Assert.assertEquals(n4, greatestOf.get(0));
    Assert.assertEquals(n3, greatestOf.get(1));
    Assert.assertEquals(n1, greatestOf.get(2));
}

@Test
public void leastOf() {
    NumbersObject n1 = new NumbersObject(1, 2, 3);
    NumbersObject n2 = new NumbersObject(2, 2, 4);
    NumbersObject n3 = new NumbersObject(-1, 2, -3);
    NumbersObject n4 = new NumbersObject(-2, 3, 5);
    NumbersObject nullNumber = new NumbersObject(null,null,null);
    List<NumbersObject> numbers = Arrays.asList(n1, n2, n3, n4, nullNumber);
    List<NumbersObject> leastOf = Ordering.natural().leastOf(numbers,3);
    Assert.assertEquals(nullNumber, leastOf.get(0));
    Assert.assertEquals(n2, leastOf.get(1));
    Assert.assertEquals(n1, leastOf.get(2));
}

private final class NumbersObject implements Comparable<NumbersObject> {

    public final Integer a;
    public final Integer b;
    public final Integer c;

    public NumbersObject(Integer a, Integer b, Integer c) {
        this.a = a;
        this.b = b;
        this.c = c;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("a",a)
                .add("b",b)
                .add("c",c)
                .toString();
    }

    @Override
    public int compareTo(NumbersObject o) {
        if (a == null && o.a == null) {
            return 0;
        } else if (a == null) {
            return -1;
        } else if (o.a == null) {
            return 1;
        } else {
            return Ints.compare(o.a,a);
        }
    }
}

ImmutableXXX

沒有 ImmutableSet 的時候會用 Collections.unmodifiableSet, 差別是 ImmutableSet 有很多好用的 api 讓程式簡單很多.
另外 ImmutableXXX 在能夠避免線性成長的 copy time 時候會盡量避免. 例如 ImmutableList.copyOf(ImmutableSet)
@Test(expected = UnsupportedOperationException.class)
public void of() {
    ImmutableSet.of("a","b").add("a");
}

@Test(expected = UnsupportedOperationException.class)
public void copyOf() {
    ImmutableSet.copyOf(Arrays.asList("a","b","c")).add("");
}

@Test
public void listCopyOfSet() {
    ImmutableList immutableList = ImmutableList.copyOf(ImmutableSet.of("a","b"));
}

Multipleset

像是 set 一樣, 差別是一個同樣的資料可以存多筆. 也可以記錄某種資料總共有幾筆.
@RunWith(Parameterized.class)
public class MultisetTest {

    @Parameterized.Parameters
    public static Collection<Object[]> data() {
        return Arrays.asList(new Object[][] {
                { HashMultiset.create() },
                {TreeMultiset.create() },
                {LinkedHashMultiset.create() },
                {ConcurrentHashMultiset.create()},
        });
    }

    private final Multiset<String> testee;

    public MultisetTest(Multiset<String> testee) {
        this.testee = testee;
    }

    @Test
    public void test() {
        testee.addAll(Arrays.asList("A","B","A"));
        Assert.assertEquals(2, testee.count("A"));
        Assert.assertEquals(1, testee.count("B"));
    }

}

public class ImmutableMultisetTest {

    @Test
    public void count() {
        ImmutableMultiset set = ImmutableMultiset.of("A","B","A");
        Assert.assertEquals(2,set.count("A"));
        Assert.assertEquals(1,set.count("B"));
    }

}

function 參數先排位置再對參數名稱

如果沒有指定要丟的參數名稱, python 就會照順序排. 如果順序對不起來就會 error.
>>> def f(a,b,c):
 print 'a:', a, 'b:', b, 'c:', c

 
>>> f(1,2,3)
a: 1 b: 2 c: 3
>>> f(1,c=3,b=2)
a: 1 b: 2 c: 3
>>> f(1,2,c=3)
a: 1 b: 2 c: 3
>>> f(1,2,b=2) # 第二個參數已經指定了, 所以就不能再指定 b

Traceback (most recent call last):
  File "", line 1, in 
    f(1,2,b=2)
TypeError: f() got multiple values for keyword argument 'b'

>>> f(c=3,2,a=1)
SyntaxError: non-keyword arg after keyword arg

>>> f(a=1,2,3)
SyntaxError: non-keyword arg after keyword arg

字串 format

帶入 dict 的值, 如果用 format api, dictionary 要帶 ** 去展開 key-value
>>> m = {'a':1.1234, "b":"qq"}
>>> '%(a)1.1f %(a)s %(b)s' %m
'1.1 1.1234 qq'


>>> '{a:1.1f} {a:s} {b:s}'.format(m) # illegal, must use **m

Traceback (most recent call last):
  File "", line 1, in 
    '{a:1.1f} {a:s} {b:s}'.format(m)
KeyError: 'a'

>>> '{a:1.1f} {a:s} {b:s}'.format(**m) # illegal, a is float type

Traceback (most recent call last):
  File "", line 1, in 
    '{a:1.1f} {a:s} {b:s}'.format(**m)
ValueError: Unknown format code 's' for object of type 'float'

>>> '{a:1.1f} {b:s}'.format(**m)
'1.1 qq'

tuple 就是不可變的 list

tuple 就是不可變的 list.
>>> list = ['c',[1,2,3]]
>>> tuple = ('c',(1,2,3))
>>> list[0]
'c'
>>> list[1]
[1, 2, 3]
>>> tuple[0]
'c'
>>> tuple[1]
(1, 2, 3)
>>> list[0] = 'q'
>>> list[0]
'q'
>>> list[1]
[1, 2, 3]
>>> tuple[0] = 'q'

Traceback (most recent call last):
  File "", line 1, in 
    tuple[0] = 'q'
TypeError: 'tuple' object does not support item assignment

function is an object

python 的 function 是一個物件, 所以可以放進 collection 中等以後被呼叫.
>>> def a():
 print 'a';

 
>>> def b():
 print 'b';

 
>>> def c():
 print 'c';

 
>>> list = [a,b,c];
>>> for f in list:
 f();

 
a
b
c

split large file

I use Notepad++ to check log and debug.
When log file is large, I need to split large file to smaller ones.
I can't find tool so write codes to split file in Java every time.
I use Python to write a tool to split file when studying python.
Welcome to use this tool if you also need to split large log file.

Lessons Learned While Benchmarking vLLM with GPU

Recently, I benchmarked vLLM on a GPU to better understand how much throughput can realistically be expected in an LLM serving setup. One ...