这几个方法用于决定上游stream的数据如何分发给下游的各个channel。

总览

rebalance Round-Robin往下游channel发送。
rescale Round-Robin往下游channel发送。
跟rebalance的不同点在于，rebalance会将数据发送发送给所有下游任务，而 rescale只将数据发送给部分下游任务。
比如上游stream的并行度为2，下游的并行度为4，那么上游1号channel中的数据会发送给下游的某两个channel，上游2号channel中的数据会发送给下游的另外两个channel。
shuffle 随机选择一个下游channel发送。
keyedBy 计算key的hashCode；再用murmurHash计算hash值，保证进一步散列；hash % maxParallelism得到下游channel的编号。

上代码，一目了然

rebalance

Round-Robin往下游channel发送。

  1. public class RebalancePartitioner<T> extends StreamPartitioner<T> {
  2.     private static final long serialVersionUID = 1L;
  4.     private int nextChannelToSendTo;
  6.     @Override
  7.     public void setup(int numberOfChannels) {
  8.         super.setup(numberOfChannels);
  10.         nextChannelToSendTo = ThreadLocalRandom.current().nextInt(numberOfChannels);
  11.     }
  13.     @Override
  14.     public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
  15.         nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels;
  16.         return nextChannelToSendTo;
  17.     }
  18. }

rescale

Round-Robin往下游channel发送。
跟rebalance的不同点在于，rebalance会将数据发送发送给所有下游任务，而 rescale只将数据发送给部分下游任务。
// TODO: 这块没怎么看懂，跟上面有什么区别，怎么就选择部分下游了？

  1. public class RescalePartitioner<T> extends StreamPartitioner<T> {
  2.     private static final long serialVersionUID = 1L;
  4.     private int nextChannelToSendTo = -1;
  6.     @Override
  7.     public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
  8.         if (++nextChannelToSendTo >= numberOfChannels) {
  9.             nextChannelToSendTo = 0;
  10.         }
  11.         return nextChannelToSendTo;
  12.     }
  13. }

shuffle

随机选择一个下游channel发送。

  1. public class ShufflePartitioner<T> extends StreamPartitioner<T> {
  2.     private static final long serialVersionUID = 1L;
  4.     private Random random = new Random();
  6.     @Override
  7.     public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
  8.         return random.nextInt(numberOfChannels);
  9.     }
  10. }

keyedBy

计算key的hashCode；再用murmurHash计算hash值，保证进一步散列；hash % maxParallelism得到下游channel的编号。

  1. public class KeyGroupStreamPartitioner<T, K> extends StreamPartitioner<T> implements ConfigurableStreamPartitioner {
  2.     private static final long serialVersionUID = 1L;
  4.     private final KeySelector<T, K> keySelector;
  6.     private int maxParallelism;
  8.     @Override
  9.     public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
  10.         K key;
  11.         try {
  12.             key = keySelector.getKey(record.getInstance().getValue());
  13.         } catch (Exception e) {
  14.             throw new RuntimeException("Could not extract key from " + record.getInstance().getValue(), e);
  15.         }
  16.         return KeyGroupRangeAssignment.assignKeyToParallelOperator(key, maxParallelism, numberOfChannels);
  17.     }
  18. }
  21. public final class KeyGroupRangeAssignment {
  23.     /**
  24.      * Assigns the given key to a parallel operator index.
  25.      * @return the index of the parallel operator to which the given key should be routed.
  26.      */
  27.     public static int assignKeyToParallelOperator(Object key, int maxParallelism, int parallelism) {
  28.         Preconditions.checkNotNull(key, "Assigned key must not be null!");
  29.         return computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
  30.     }
  32.     /**
  33.      * Assigns the given key to a key-group index.
  34.      * @return the key-group to which the given key is assigned
  35.      */
  36.     public static int assignToKeyGroup(Object key, int maxParallelism) {
  37.         Preconditions.checkNotNull(key, "Assigned key must not be null!");
  38.         return computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
  39.     }
  41.     /**
  42.      * Assigns the given key to a key-group index.
  43.      * @return the key-group to which the given key is assigned
  44.      */
  45.     public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {
  46.         return MathUtils.murmurHash(keyHash) % maxParallelism;
  47.     }
  48. }