Focus on Java

2018-07-08-Java逆变与协变

发表于 2018-07-08 |

引子

《Effective Java》中第25条中《列表优于数组》中提到数组是协变的，相反泛型是不可变的

其实用于描述Java类型转换后的继承关系一共有三种，协变，逆变，不可变

其定义为：

如果A、B表示类型 f(⋅) 表示类型转换，≤ 表示继承关系（比如，A≤B 表示A是由B派生出来的子类）；

f(⋅) 是逆变（contravariant）的，当A≤B时有f(B)≤f(A)成立
f(⋅) 是协变（covariant）的，当A≤B时有f(A)≤f(B)成立
f(⋅) 是不变（invariant）的，当A≤B时上述两个式子均不成立，即f(A)与f(B)相互之间没有继承关系

协变

数组是协变的，那就意味着String是Object的子类，则String[] 是 Object[]的子类，但是会有一个问题：

1 2	Object[] objArray = new Integer[1]; objArray[0] = "a string";

这段代码是合法的，但是在运行时就会因为类型不符报错

不可变

泛型是不可变的，这意味着

1 2	ArrayList<Object> objArray = new ArrayList<Object>(); objArray.add("a string");

是无法通过编译的。根据不可变的定义，

和 ArrayList没有继承关系```


这样的设计是为了保证类型安全，根据《Effective Java》中的说法：

```java

    // Why generic array creation is illegal - won't compile
    List<String>[] stringLists = new ArrayList<String>[1]; // (1)
    List<Integer> intList = Arrays.asList(42);             // (2)
    Object[] objects = stringLists;                        // (3)
    objects[0] = intList;                                  // (4)
    String s = stringLists[0].get(0);                      // (5)

假设(1)是编译正确的，那么在(5)的时候就必然会出现类型不匹配，因为它尝试把整型赋值给字符类型的

泛型类型中利用通配符(extends/super)来实现协变和逆变

List<? extends Fruit> 表明每个item是Fruit/Fruit的子类，这其实表明了泛型的上线，实现了协变

同样，List<? super Fruit> 表明每个item都是Fruit/Furit的基类，这表明了泛型的下线，实现了逆变

泛型的协变/逆变使用依靠着一个PECS原则，即Provider Extends Consumer Super

还以List为例：

1
2
3

// 前提为Apple为Fruit的派生子类
List<? extends Fruit> list = new ArrayList<Apple>();
list.add(new Apple());

这样的写法是无法通过编译的，会提示类型不符，因为? extends Furit表明了Furit或者Furit的派生子类。如果我们存入Apple，在get时强转为Apple自然是不会有问题，但是如果我存入Banana，同样是水果，那么在get时就会报错，这样是类型不安全的

泛型通过擦除来实现的，? extends Furit在编译阶段只是一个标记，和数组具体化类型是不一样的

那么作为一个Consumer，应当使用super

1
2
3

// 前提为Apple为Fruit的派生子类
 List<? super Fruit> list = new ArrayList<Food>();
 list.add(new Apple());

这段代码是编译通过的，只要类型要求是Fruit/Fruit的基类，那么存入的类型必定可以强转为Fruit/Fruit的基类，是类型安全的

利用MDC + dubbo filter记录dubbo日志

发表于 2018-05-01 |

MDC

MDC Mapped Diagnostic Contexts 是一个诊断的工具类。支持logback log4j，提供的方法很简单，get put remove clear，这些方法针是针对线程的上下文的操作，其实MDC的实现也比较简单，核心是对ThreadLocal的一个子类InheritableThreadLocal的操作，这样保证了该类的子类共享该类的上下文

dubbo filter

dubbo filter 可以为dubbo服务传递上下文，这样结合MDC就可以用来分析dubbo服务上下文之间的调用，特别在并发量大的时候能够很清晰的grep出来同一个请求的上下文，让定位问题更简单

例子

logback.xml

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{HH:mm:ss.SSS} [%thread][%X{trace_id}] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>
    <root level="INFO">
        <appender-ref ref="STDOUT"/>
    </root>
</configuration>

注意%X{trace_id}记录了trace_id的变量值

关于logback其他参数

%msg就是单纯的日志内容

%n 换行

%-5level 左对齐最短长度5 日志等级

打印日志也是对性能有一定的影响的，打印的内容需要注意下

dubbo filter

public class TraceFilter implements Filter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // trace_id 可以从请求参数中获取
        MDC.put("trace_id", UUID.randomUUID().toString());
        return invoker.invoke(invocation);
    }
}

在请求开始时记录下来 trace_id，也可以在请求头中加入trace相关信息，这样在分布式系统中就可以将日志信息串联起来

Java设计模式之观察者模式

发表于 2018-03-18 |

观察者模式是一种比较经典的模式。被观察者可以对应多个观察者，是一种一对多的模式。被观察者一旦有变动，就会通知到所有注册的观察者，被观察者可以增加也可以移除，易于扩展。观察者模式又被称之为发布订阅模式，监听器模式等等，下面通过一个买票的例子来介绍一下这种模式

买票

买票，这个动作隐含了两个主体，一个是人，一个是票，同时这个动作还包含了一个隐藏的动作，通知。好了，这个通知就是我们用观察者模式的地方了。

大麦网有这么个功能，缺货登记，就是有票了就告诉登记的人。那我们改造一下，变成票有变化，就通知登记人。

上面的例子可以很明显分辨出来，票就是被观察的对象，人是观察者。

被观察者

下一步，来梳理一下通知这部分需要哪些功能。

首先，肯定是我订阅这个票的动态，暂时取名为addObserver，其次有订阅动态就有取消订阅removeObserver，当然最重要的还是通知notifyObservers。

下面我们就定义一个抽象类Observable—实现上面所提到的三个基本功能

package me.learn.pattern.observer;

import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;

public abstract class Observable {

    private List<Observer> observers = new LinkedList<>();

    protected void addObserver(Observer observer) {
        this.observers.add(observer);
    }

    protected void addObservers(Observer... observers) {
        this.observers.addAll(Arrays.asList(observers));
    }

    protected void removeObserver(Observer observer) {
        this.observers.remove(observer);
    }

    protected void removeAllObservers() {
        this.observers.clear();
    }

    protected void notifyObservers(Object data) {
        for (Observer observer : observers) {
            observer.update(this, data);
        }
    }

}

观察者

买票的另一主体是人。那么一个人在这个过程中他想要的其实很简单，就是通知到我，就像大麦的缺货登记，我登记了，你要通知我，也就是调用我的方法，然后我可以做我收到这个通知之后的事情。那就很明确了，观察者需要的就是一个类似更新update的方法。我们定义一下这个观察者的接口。

package me.learn.pattern.observer;

public interface Observer {
    void update(Observable observable, Object data);
}

接下来只要定义票继承被观察者基类，人实现观察者方法就可以了。

再说买票

其实，上面的过程已经说完了观察者模式，但是让我们再想想缺货登记。票和人，其实都既是观察者也是被观察者，票需要被观察余量，也需要观察人的购票；人既需要观察票的动态，也是被观察是否购票的对象。所以我们来写一个既是观察者也是被观察者的例子吧。

首先定义票，继承被观察者，并实现观察者。

package me.learn.pattern.observer;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

public class Ticket extends Observable implements Observer {

    private AtomicInteger left;

    public Ticket() {
        this.left = new AtomicInteger(0);
    }

    public Ticket(int left) {
        this.left = new AtomicInteger(left);
    }

    public synchronized void setLeft(int left) {
        this.left.set(left);
    }

    public synchronized void sold(int number) {
        if (this.left.get() >= number) {
            this.left.compareAndSet(this.left.get(), this.left.intValue() - number);
            super.notifyObservers(this.left.get());
        } else {
            System.out.println("sold out");
        }
    }

    public synchronized void soldAllTickets() {
        this.left.set(0);
        super.notifyObservers(this.left.get());
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Ticket ticket = (Ticket) o;
        return Objects.equals(left, ticket.left);
    }

    @Override
    public int hashCode() {
        return Objects.hash(left);
    }

    @Override
    public String toString() {
        return "Ticket";
    }

    @Override
    public void update(Observable observable, Object data) {
        System.out.println(observable + " buys "+data+" ticket");
        sold((Integer) data);
    }
}

其次，购票人。

package me.learn.pattern.observer;

public class PersonWhoWantTickets extends Observable implements Observer {

    private String name;

    public PersonWhoWantTickets(String name) {
        this.name = name;
    }

    public void buyOneTicket() {
        System.out.println(name + " buys one ticket");
        super.notifyObservers(1);
    }

    @Override
    public void update(Observable observable, Object data) {
        System.out.println(this + " received notice# tickets left["+data+"]");
        if ((int)data == 0) {
            System.out.println("oh, i can not buy more");
        } else {
            buyOneTicket();
        }
    }

    @Override
    public String toString() {
        return "PersonWhoWantTickets{" +
                "name='" + name + '\'' +
                '}';
    }

    public static void main(String[] argv) {
        Ticket ticket = new Ticket(13);

        PersonWhoWantTickets person = new PersonWhoWantTickets("person");
        person.addObserver(ticket);
        ticket.addObservers(person);
        ticket.sold(5);

        person.buyOneTicket();
    }
}

有意思，这样就有意思了。

BlockingQueue实现生产者消费者模型

发表于 2018-03-06 |

package me.learn.treads;

import java.util.concurrent.BlockingQueue;

public interface Consumer extends Runnable {
    void consume(BlockingQueue queue) throws Exception;
}

package me.learn.treads;

import java.util.concurrent.BlockingQueue;

public interface Producer extends Runnable {
    void produce(BlockingQueue queue) throws Exception;
}

package me.learn.treads;

public abstract class Task {
    protected abstract void run() throws Exception;
}

package me.learn.treads;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public abstract class AbstractConsumer implements Consumer {

    protected abstract BlockingQueue attachQueue();
    protected final AtomicInteger increaseInt = new AtomicInteger();
    @Override
    public void run() {
        Thread.currentThread().setName("consumer#" + increaseInt.incrementAndGet());
        System.out.println("consumer at thread$"+Thread.currentThread().getName());
        while (true) {
            try {
                consume(attachQueue());
            } catch (Exception e) {
                e.printStackTrace();
                break;
            }
        }
    }
}

package me.learn.treads;

import java.util.concurrent.BlockingQueue;

public abstract class AbstractProducer implements Producer {

    protected abstract BlockingQueue attachQueue();

    @Override
    public void run() {
        System.out.println("producer at Thread#"+Thread.currentThread().getName());
        while (true) {
            try {
                produce(attachQueue());
            } catch (Exception e) {
                e.printStackTrace();
                break;
            }
        }
    }
}

package me.learn.treads;

import java.util.concurrent.BlockingQueue;

public class MyConsumer extends AbstractConsumer {

    private BlockingQueue queue;
    public MyConsumer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    protected BlockingQueue attachQueue() {
        return queue;
    }

    @Override
    public void consume(BlockingQueue queue) throws Exception {
        BlockQueueProducerConsumerModel.MyTask task = (BlockQueueProducerConsumerModel.MyTask) queue.take();
        if (task != null) {
            System.out.print(Thread.currentThread().getName());
            task.run();
        }
    }
}

package me.learn.treads;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class MyProducer extends AbstractProducer {

    private final AtomicInteger increaseNumber = new AtomicInteger();

    private BlockingQueue queue;

    private final Random random = new Random();

    public MyProducer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    protected BlockingQueue attachQueue() {
        return queue;
    }

    @Override
    public void produce(BlockingQueue queue) throws Exception {
        queue.offer(BlockQueueProducerConsumerModel.MyTask.newTask(() -> increaseNumber.incrementAndGet()));
        System.out.println("queue size " + queue.size());
        Thread.sleep(500);
    }

    private int getRandomMillisecond() {
        return random.ints(0,5).findAny().getAsInt() * 1000;
    }
}

package me.learn.treads;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

public final class BlockQueueProducerConsumerModel {
    private static final BlockingQueue<Task> myBlockQueue = new LinkedBlockingQueue<>();
    private final AbstractProducer producer;
    private final AbstractConsumer consumer;

    public static final AtomicInteger increaseInt = new AtomicInteger();

    private static final Random random = new Random();

    public BlockQueueProducerConsumerModel(AbstractProducer producer, AbstractConsumer consumer) {
        this.producer = producer;
        this.consumer = consumer;
    }

    public void start() {
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.execute(producer);
        executorService.execute(consumer);
        executorService.execute(consumer);
        executorService.execute(consumer);
        executorService.execute(consumer);
        executorService.execute(consumer);
        executorService.execute(consumer);
    }

    public static void main(String[] argv) {
        BlockQueueProducerConsumerModel model = new BlockQueueProducerConsumerModel(
                new MyProducer(myBlockQueue),
                new MyConsumer(myBlockQueue));
        model.start();
    }

    static class MyTask extends Task {

        private Supplier supplier;

        private MyTask(Supplier supplier) {
            this.supplier = supplier;
        }

        public static MyTask newTask(Supplier supplier) {
            return new MyTask(supplier);
        }

        @Override
        protected void run() throws Exception {
            System.out.println("running #"+this.supplier.get());
            Thread.sleep(getRandomMillisecond());
        }

        private int getRandomMillisecond() {
            return random.ints(0,5).findAny().getAsInt() * 1000;
        }
    }
}

ForkJoinPool实现快排

发表于 2018-02-27 |

ForkJoin实现快速排序

package me.learn.treads;

import java.util.Arrays;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

public class FastSort<T extends Comparable<T>> extends RecursiveAction {
    private final T[] arrays;
    private int low;
    private int high;

    public FastSort(T[] arrays) {
        this.arrays = arrays;
        this.low = 0;
        this.high = this.arrays.length-1;
    }

    public FastSort(T[] arrays, int low, int high) {
        System.out.println(Arrays.toString(arrays));
        this.arrays = arrays;
        this.low = low;
        this.high = high;
    }

    public static <T extends Comparable<T>> int partition(T[] arrays, int left, int right) {
        int k = left;
        T pivot = arrays[right];
        for (int i=left; i<right; ++i)
            if (arrays[i].compareTo(pivot) < 0) swap(arrays, i, k++);
        swap(arrays, k, right);
        return k;
    }

    static <T extends Comparable<T>> void swap(T[] arrays, int i, int j) {
        if (i != j) {
            T tmp = arrays[i];
            arrays[i] = arrays[j];
            arrays[j] = tmp;
        }
    }

    @Override
    protected void compute() {
        if (low < high) {
            int pivot = partition(arrays, low, high);
            FastSort<T> fastSort1 = new FastSort<>(arrays, low, pivot-1);
            FastSort<T> fastSort2 = new FastSort<>(arrays, pivot+1, high);
            invokeAll(fastSort1, fastSort2);

        }
    }

    public static void main(String[] argv) throws InterruptedException {
        ForkJoinPool pool = ForkJoinPool.commonPool();
        Integer[] arr = new Integer[]{3,7,8,5,2,1,9,5,4};
        FastSort<Integer> fastSort = new FastSort<>(arr);
        pool.submit(fastSort);
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println(Arrays.toString(arr));
    }
}