Netty编程（十）—— 参数优化

发表于 2021-11-29

这是我参与11月更文挑战的第28天，活动详情查看：2021最后一次更文挑战

CONNECT_TIMEOUT_MILLIS

属于 SocketChannal 的参数
用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常
注意：Netty 中不要用成了SO_TIMEOUT 主要用在阻塞 IO，而 Netty 是非阻塞 IO

使用

java复制代码public class TestParam {
    public static void main(String[] args) {
        // SocketChannel 5s内未建立连接就抛出异常
        new Bootstrap().option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
        
        // ServerSocketChannel 5s内未建立连接就抛出异常
        new ServerBootstrap().option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000);
        // SocketChannel 5s内未建立连接就抛出异常
        new ServerBootstrap().childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
    }
}

客户端通过 Bootstrap.option 函数来配置参数，配置参数作用于 SocketChannel
服务器通过

1	erlang复制代码ServerBootstrap.

来配置参数，但是对于不同的 Channel 需要选择不同的方法

+ 通过 `option` 来配置 **ServerSocketChannel** 上的参数
+ 通过 `childOption` 来配置 **SocketChannel** 上的参数

源码分析

客户端中连接服务器的线程是 NIO 线程，抛出异常的是主线程。这是如何做到超时判断以及线程通信的呢？

AbstractNioChannel.AbstractNioUnsafe.connect方法中

java复制代码public final void connect(
                final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    
    ...
        
    // Schedule connect timeout.
    // 设置超时时间，通过option方法传入的CONNECT_TIMEOUT_MILLIS参数进行设置
    int connectTimeoutMillis = config().getConnectTimeoutMillis();
    // 如果超时时间大于0
    if (connectTimeoutMillis > 0) {
        // 创建一个定时任务，延时connectTimeoutMillis（设置的超时时间时间）后执行
        // schedule(Runnable command, long delay, TimeUnit unit)
        connectTimeoutFuture = eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                // 判断是否建立连接，Promise进行NIO线程与主线程之间的通信
                // 如果超时，则通过tryFailure方法将异常放入Promise中
                // 在主线程中抛出
                ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                ConnectTimeoutException cause = new ConnectTimeoutException("connection timed out: " + remoteAddress);
                if (connectPromise != null && connectPromise.tryFailure(cause)) {
                    close(voidPromise());
                }
            }
        }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
    }
    
   	...
        
}

超时的判断主要是通过 Eventloop 的 schedule 方法和 Promise 共同实现的

schedule 设置了一个定时任务，延迟connectTimeoutMillis秒后执行该方法
如果指定时间内没有建立连接，则会执行其中的任务
- 任务负责创建 ConnectTimeoutException 异常，并将异常通过 Pormise 传给主线程并抛出

SO_BACKLOG

该参数是 ServerSocketChannel 的参数，在服务器端

三次握手与连接队列

第一次握手时，因为客户端与服务器之间的连接还未完全建立，与这个客户端的连接信息会被放入半连接队列中

当完成三次握手以后，连接会被放入全连接队列中

服务器处理Accept事件是在TCP三次握手，也就是建立连接之后。服务器会从全连接队列中获取连接并进行处理

在 linux 2.2 之前，backlog 大小包括了两个队列的大小，在 linux 2.2 之后，分别用下面两个参数来控制

半连接队列 - sync queue
- 大小通过 /proc/sys/net/ipv4/tcp_max_syn_backlog 指定，在 syncookies 启用的情况下，逻辑上没有最大值限制，这个设置便被忽略
全连接队列 - accept queue
- 其大小通过 /proc/sys/net/core/somaxconn 指定，在使用 listen 函数时，内核会根据传入的 backlog 参数与系统参数，取二者的较小值
- 如果 accpet queue 队列满了，server 将发送一个拒绝连接的错误信息到 client

作用

在Netty中，SO_BACKLOG主要用于设置全连接队列的大小。当处理Accept的速率小于连接建立的速率时，全连接队列中堆积的连接数大于SO_BACKLOG设置的值是，便会抛出异常

设置方式如下

1 2	scss复制代码// 设置全连接队列，大小为2 new ServerBootstrap().option(ChannelOption.SO_BACKLOG, 2);

默认值

backlog参数在NioSocketChannel.doBind方法被使用

java复制代码@Override
protected void doBind(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        javaChannel().bind(localAddress, config.getBacklog());
    } else {
        javaChannel().socket().bind(localAddress, config.getBacklog());
    }
}

其中backlog被保存在了DefaultServerSocketChannelConfig配置类中

1	arduino复制代码private volatile int backlog = NetUtil.SOMAXCONN;

具体的赋值操作如下

java复制代码SOMAXCONN = AccessController.doPrivileged(new PrivilegedAction<Integer>() {
    @Override
    public Integer run() {
        // Determine the default somaxconn (server socket backlog) value of the platform.
        // The known defaults:
        // - Windows NT Server 4.0+: 200
        // - Linux and Mac OS X: 128
        int somaxconn = PlatformDependent.isWindows() ? 200 : 128;
        File file = new File("/proc/sys/net/core/somaxconn");
        BufferedReader in = null;
        try {
            // file.exists() may throw a SecurityException if a SecurityManager is used, so execute it in the
            // try / catch block.
            // See https://github.com/netty/netty/issues/4936
            if (file.exists()) {
                in = new BufferedReader(new FileReader(file));
                // 将somaxconn设置为Linux配置文件中设置的值
                somaxconn = Integer.parseInt(in.readLine());
                if (logger.isDebugEnabled()) {
                    logger.debug("{}: {}", file, somaxconn);
                }
            } else {
                ...
            }
            ...
        }  
        // 返回backlog的值
        return somaxconn;
    }
}

backlog的值会根据操作系统的不同，来

选择不同的默认值

+ Windows 200
+ Linux/Mac OS 128

如果配置文件/proc/sys/net/core/somaxconn存在，会读取配置文件中的值，并将backlog的值设置为配置文件中指定的

TCP_NODELAY

属于 SocketChannal 参数
因为 Nagle 算法，数据包会堆积到一定的数量后一起发送，这就可能导致数据的发送存在一定的延时
该参数默认为false，如果不希望的发送被延时，则需要将该值设置为true

SO_SNDBUF & SO_RCVBUF

SO_SNDBUF 属于 SocketChannal 参数
SO_RCVBUF 既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）
该参数用于指定接收方与发送方的滑动窗口大小，不过不建议去调整这两个参数，也不应该过大，否则会占用操作系统的缓冲区

ALLOCATOR

属于 SocketChannal 参数
用来配置 ByteBuf 是池化还是非池化，是直接内存还是堆内存

使用

1
2
3

scss复制代码// 选择ALLOCATOR参数，设置SocketChannel中分配的ByteBuf类型
// 第二个参数需要传入一个ByteBufAllocator，用于指定生成的 ByteBuf 的类型
new ServerBootstrap().childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator());

ByteBufAllocator类型

池化并使用直接内存

1 2	arduino复制代码// true表示使用直接内存 new PooledByteBufAllocator(true);

池化并使用堆内存

1 2	arduino复制代码// false表示使用堆内存 new PooledByteBufAllocator(false);

非池化并使用直接内存

1 2	arduino复制代码// ture表示使用直接内存 new UnpooledByteBufAllocator(true);

非池化并使用堆内存

1 2	arduino复制代码// false表示使用堆内存 new UnpooledByteBufAllocator(false);

RCVBUF_ALLOCATOR

属于 SocketChannal 参数
控制 Netty 接收缓冲区大小
负责入站数据的分配，决定入站缓冲区的大小（并可动态调整），统一采用 direct 直接内存，具体池化还是非池化由 allocator 决定

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

【API】微信支付简单集成

发表于 2021-11-29

这是我参与11月更文挑战的第13天，活动详情查看：2021最后一次更文挑战

说明

当服务服务中需要支付功能，可以尝试使用微信支付。微信支付提供企业向用户付款的功能，支持企业通过API接口付款，或通过微信支付商户平台网页功能操作付款。

功能开通

首先需要在官网进行开通，地址： #PC网站接入支付

授权说明

通过网页授权获取用户的openid（企业向微信用户个人付款目前支持向指定微信用户的openid付款。），具体获取方法可查网页授权

API调用代码

请求示例

只写了些简单参数，具体参数详情见【微信支付】付款开发者文档

java复制代码Map<String, String> reqData = new HashMap<>();
// 签名
reqData.put("sign","abc");
// 用户openid
reqData.put("openid","1657446542123");
// 校验用户姓名选项
reqData.put("check_name","NO_CHECK");
// 付款金额，单位为分
reqData.put("amount","30.0");
// 付款备注
reqData.put("desc","付款");
// 商户账号appid
reqData.put("mch_appid", "wx1234567065d35555");
// 商户号
reqData.put("mchid", "1768826471");
// Ip地址
reqData.put("spbill_create_ip", InetAddress.getLocalHost().getHostAddress());
// 随机字符串
reqData.put("nonce_str", "lst");
// 转成xml格式
String xml = mapToXml(reqData);
String url = "https://api.mch.weixin.qq.com/mmpaymkttransfers/promotion/transfers";
// 发起请求
String response = HttpService.Post(xml, url, true, 5000);

map转xml方法

java复制代码public static String mapToXml(Map<String, String> data) throws Exception {
        Document document = PayXmlUtil.newDocument();
        Element root = document.createElement("xml");
        document.appendChild(root);
        Iterator var3 = data.keySet().iterator();

        while(var3.hasNext()) {
            String key = (String)var3.next();
            String value = (String)data.get(key);
            if (value == null) {
                value = "";
            }

            value = value.trim();
            Element filed = document.createElement(key);
            filed.appendChild(document.createTextNode(value));
            root.appendChild(filed);
        }

        TransformerFactory tf = TransformerFactory.newInstance();
        Transformer transformer = tf.newTransformer();
        DOMSource source = new DOMSource(document);
        transformer.setOutputProperty("encoding", "UTF-8");
        transformer.setOutputProperty("indent", "yes");
        StringWriter writer = new StringWriter();
        StreamResult result = new StreamResult(writer);
        transformer.transform(source, result);
        String output = writer.getBuffer().toString();

        try {
            writer.close();
            return output;
        } catch (Exception e) {
            log.error(e.toString());
        }
}

注意

请求头中需要设置格式为xml

1	java复制代码httpPost.addHeader("Content-Type", "text/xml");

参考

【微信支付】付款开发者文档

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Synchronized原理分析文章简介内容导航什么时

发表于 2021-11-29

「这是我参与11月更文挑战的第29天，活动详情查看：2021最后一次更文挑战」

文章简介

synchronized想必大家都不陌生，用来解决线程安全问题的利器。同时也是Java高级程序员面试比较常见的面试题。这篇文正会带大家彻底了解synchronized的实现。

内容导航

什么时候需要用Synchronized
synchronized的使用
synchronized的实现原理分析

什么时候需要用Synchronized

想必大家对synchronized都不陌生，主要作用是在多个线程操作共享数据的时候，保证对共享数据访问的线程安全性。

比如在下面这个图片中，两个线程对于i这个共享变量同时做i++递增操作，那么这个时候对于i这个值来说就存在一个不确定性，也就是说理论上i的值应该是2，但是也可能是1。而导致这个问题的原因是线程并行执行i++操作并不是原子的，存在线程安全问题。所以通常来说解决办法是通过加锁来实现线程的串行执行，而synchronized就是java中锁的实现的关键字。

synchronized在并发编程中是一个非常重要的角色，在JDK1.6之前，它是一个重量级锁的角色，但是在JDK1.6之后对synchronized做了优化，优化以后性能有了较大的提升(这块会在后面做详细的分析)。

先来看一下synchronized的使用

Synchronized的使用

synchronized有三种使用方法，这三种使用方法分别对应三种不同的作用域，代码如下

修饰普通同步方法

将synchronized修饰在普通同步方法，那么该锁的作用域是在当前实例对象范围内,也就是说对于 SyncDemosd=newSyncDemo();这一个实例对象sd来说，多个线程访问access方法会有锁的限制。如果access已经有线程持有了锁，那这个线程会独占锁，直到锁释放完毕之前，其他线程都会被阻塞

java复制代码public SyncDemo{
   Object lock =new Object();
    //形式1
    public synchronized void access(){
       //
    }
    //形式2,作用域等同于形式1
    public void access1(){
       synchronized(lock){
         //
       }
    }
    //形式3，作用域等同于前面两种
    public void access2(){
       synchronized(this){
          //
       }
    }
}

修饰静态同步方法

修饰静态同步方法或者静态对象、类，那么这个锁的作用范围是类级别。举个简单的例子，

1 2	java复制代码SyncDemo sd=SyncDemo(); SyncDemo sd2=new SyncDemo();}

两个不同的实例sd和sd2，如果sd这个实例访问access方法并且成功持有了锁，那么sd2这个对象如果同样来访问access方法，那么它必须要等待sd这个对象的锁释放以后，sd2这个对象的线程才能访问该方法，这就是类锁；也就是说类锁就相当于全局锁的概念，作用范围是类级别。

这里抛一个小问题，大家看看能不能回答，如果不能也没关系，后面会讲解；问题是如果sd先访问access获得了锁，sd2对象的线程再访问access1方法，那么它会被阻塞吗？

java复制代码public SyncDemo{
    static Object lock=new Object();
    //形式1
    public synchronized static void access(){
       //
    }
    //形式2等同于形式1
    public void access1(){
       synchronized(lock){
          //
       }
    }
    //形式3等同于前面两种
    public void access2(){
        synchronzied(SyncDemo.class){
          //
        }
    }
}

步方法块

java复制代码public SyncDemo{
   Object lock=new Object();
   public void access(){
       //do something
       synchronized(lock){
         //
       }
   }
}

通过演示3种不同锁的使用，让大家对synchronized有了初步的认识。当一个线程视图访问带有synchronized修饰的同步代码块或者方法时，必须要先获得锁。当方法执行完毕退出以后或者出现异常的情况下会自动释放锁。如果大家认真看了上面的三个案例，那么应该知道锁的范围控制是由对象的作用域决定的。对象的作用域越大，那么锁的范围也就越大，因此我们可以得出一个初步的猜想，synchronized和对象有非常大的关系。那么，接下来就去剖析一下锁的原理

Synchronized的实现原理分析

当一个线程尝试访问synchronized修饰的代码块时，它首先要获得锁，那么这个锁到底存在哪里呢？

对象在内存中的布局

synchronized实现的锁是存储在Java对象头里，什么是对象头呢？在Hotspot虚拟机中，对象在内存中的存储布局，可以分为三个区域:对象头(Header)、实例数据(Instance Data)、对齐填充(Padding)

当我们在Java代码中，使用new创建一个对象实例的时候，（hotspot虚拟机）JVM层面实际上会创建一个 instanceOopDesc对象。

Hotspot虚拟机采用OOP-Klass模型来描述Java对象实例，OOP(Ordinary Object Point)指的是普通对象指针，Klass用来描述对象实例的具体类型。Hotspot采用instanceOopDesc和arrayOopDesc来描述对象头，arrayOopDesc对象用来描述数组类型

instanceOopDesc的定义在Hotspot源码中的 instanceOop.hpp文件中，另外，arrayOopDesc的定义对应 arrayOop.hpp

c++复制代码class instanceOopDesc : public oopDesc {
 public:
  // aligned header size.
  static int header_size() { return sizeof(instanceOopDesc)/HeapWordSize; }
  // If compressed, the offset of the fields of the instance may not be aligned.
  static int base_offset_in_bytes() {
    // offset computation code breaks if UseCompressedClassPointers
    // only is true
    return (UseCompressedOops && UseCompressedClassPointers) ?
             klass_gap_offset_in_bytes() :
             sizeof(instanceOopDesc);
  }
  static bool contains_field_offset(int offset, int nonstatic_field_size) {
    int base_in_bytes = base_offset_in_bytes();
    return (offset >= base_in_bytes &&
            (offset-base_in_bytes) < nonstatic_field_size * heapOopSize);
  }
};
#endif // SHARE_VM_OOPS_INSTANCEOOP_HPP

从instanceOopDesc代码中可以看到 instanceOopDesc继承自oopDesc，oopDesc的定义载Hotspot源码中的 oop.hpp文件中

c++复制代码class oopDesc {
  friend class VMStructs;
 private:
  volatile markOop  _mark;
  union _metadata {
    Klass*      _klass;
    narrowKlass _compressed_klass;
  } _metadata;
  // Fast access to barrier set.  Must be initialized.
  static BarrierSet* _bs;
  ...
}

在普通实例对象中，oopDesc的定义包含两个成员，分别是 _mark和 _metadata

_mark表示对象标记、属于markOop类型，也就是接下来要讲解的Mark World，它记录了对象和锁有关的信息

_metadata表示类元信息，类元信息存储的是对象指向它的类元数据(Klass)的首地址，其中Klass表示普通指针、 _compressed_klass表示压缩类指针

Mark Word

在前面我们提到过，普通对象的对象头由两部分组成，分别是markOop以及类元信息，markOop官方称为Mark Word
在Hotspot中，markOop的定义在 markOop.hpp文件中，代码如下

c++复制代码class markOopDesc: public oopDesc {
 private:
  // Conversion
  uintptr_t value() const { return (uintptr_t) this; }
 public:
  // Constants
  enum { age_bits                 = 4,  //分代年龄
         lock_bits                = 2, //锁标识
         biased_lock_bits         = 1, //是否为偏向锁
         max_hash_bits            = BitsPerWord - age_bits - lock_bits - biased_lock_bits,
         hash_bits                = max_hash_bits > 31 ? 31 : max_hash_bits, //对象的hashcode
         cms_bits                 = LP64_ONLY(1) NOT_LP64(0),
         epoch_bits               = 2 //偏向锁的时间戳
  };
...

Mark word记录了对象和锁有关的信息，当某个对象被synchronized关键字当成同步锁时，那么围绕这个锁的一系列操作都和Mark word有关系。Mark Word在32位虚拟机的长度是32bit、在64位虚拟机的长度是64bit。
Mark Word里面存储的数据会随着锁标志位的变化而变化，Mark Word可能变化为存储以下5中情况

锁标志位的表示意义

锁标识 lock=00 表示轻量级锁
锁标识 lock=10 表示重量级锁
偏向锁标识 biased_lock=1表示偏向锁
偏向锁标识 biased_lock=0且锁标识=01表示无锁状态

到目前为止，我们再总结一下前面的内容，synchronized(lock)中的lock可以用Java中任何一个对象来表示，而锁标识的存储实际上就是在lock这个对象中的对象头内。大家懂了吗？

其实前面只提到了锁标志位的存储，但是为什么任意一个Java对象都能成为锁对象呢？

首先，Java中的每个对象都派生自Object类，而每个Java Object在JVM内部都有一个native的C++对象 oop/oopDesc进行对应。
其次，线程在获取锁的时候，实际上就是获得一个监视器对象(monitor) ,monitor可以认为是一个同步对象，所有的Java对象是天生携带monitor.
在hotspot源码的 markOop.hpp文件中，可以看到下面这段代码。

c++复制代码ObjectMonitor* monitor() const {
    assert(has_monitor(), "check");
    // Use xor instead of &~ to provide one extra tag-bit check.
    return (ObjectMonitor*) (value() ^ monitor_value);
  }

多个线程访问同步代码块时，相当于去争抢对象监视器修改对象中的锁标识,上面的代码中ObjectMonitor这个对象和线程争抢锁的逻辑有密切的关系(后续会详细分析)

锁的升级

前面提到了锁的几个概念，偏向锁、轻量级锁、重量级锁。在JDK1.6之前，synchronized是一个重量级锁，性能比较差。从JDK1.6开始，为了减少获得锁和释放锁带来的性能消耗，synchronized进行了优化，引入了偏向锁和轻量级锁的概念。所以从JDK1.6开始，锁一共会有四种状态，锁的状态根据竞争激烈程度从低到高分别是:无锁状态->偏向锁状态->轻量级锁状态->重量级锁状态。这几个状态会随着锁竞争的情况逐步升级。为了提高获得锁和释放锁的效率，锁可以升级但是不能降级。
下面就详细讲解synchronized的三种锁的状态及升级原理

偏向锁

在大多数的情况下，锁不仅不存在多线程的竞争，而且总是由同一个线程获得。因此为了让线程获得锁的代价更低引入了偏向锁的概念。偏向锁的意思是如果一个线程获得了一个偏向锁，如果在接下来的一段时间中没有其他线程来竞争锁，那么持有偏向锁的线程再次进入或者退出同一个同步代码块，不需要再次进行抢占锁和释放锁的操作。偏向锁可以通过 -XX:+UseBiasedLocking开启或者关闭

偏向锁的获取

偏向锁的获取过程非常简单，当一个线程访问同步块获取锁时，会在对象头和栈帧中的锁记录里存储偏向锁的线程ID，表示哪个线程获得了偏向锁，结合前面分析的Mark Word来分析一下偏向锁的获取逻辑

首先获取目标对象的Mark Word，根据锁的标识为和epoch去判断当前是否处于可偏向的状态
如果为可偏向状态，则通过CAS操作将自己的线程ID写入到MarkWord，如果CAS操作成功，则表示当前线程成功获取到偏向锁，继续执行同步代码块
如果是已偏向状态，先检测MarkWord中存储的threadID和当前访问的线程的threadID是否相等，如果相等，表示当前线程已经获得了偏向锁，则不需要再获得锁直接执行同步代码；如果不相等，则证明当前锁偏向于其他线程，需要撤销偏向锁。

CAS:表示自旋锁，由于线程的阻塞和唤醒需要CPU从用户态转为核心态，频繁的阻塞和唤醒对CPU来说性能开销很大。同时，很多对象锁的锁定状态指会持续很短的时间，因此引入了自旋锁，所谓自旋就是一个无意义的死循环，在循环体内不断的重行竞争锁。当然，自旋的次数会有限制，超出指定的限制会升级到阻塞锁。

偏向锁的撤销

当其他线程尝试竞争偏向锁时，持有偏向锁的线程才会释放偏向锁，撤销偏向锁的过程需要等待一个全局安全点(所有工作线程都停止字节码的执行)。

首先，暂停拥有偏向锁的线程，然后检查偏向锁的线程是否为存活状态
如果线程已经死了，直接把对象头设置为无锁状态
如果还活着，当达到全局安全点时获得偏向锁的线程会被挂起，接着偏向锁升级为轻量级锁，然后唤醒被阻塞在全局安全点的线程继续往下执行同步代码

偏向锁的获取流程图

轻量级锁

前面我们知道，当存在超过一个线程在竞争同一个同步代码块时，会发生偏向锁的撤销。偏向锁撤销以后对象会可能会处于两种状态

一种是不可偏向的无锁状态，简单来说就是已经获得偏向锁的线程已经退出了同步代码块，那么这个时候会撤销偏向锁，并升级为轻量级锁
一种是不可偏向的已锁状态，简单来说就是已经获得偏向锁的线程正在执行同步代码块，那么这个时候会升级到轻量级锁并且被原持有锁的线程获得锁

那么升级到轻量级锁以后的加锁过程和解锁过程是怎么样的呢?

轻量级锁加锁

JVM会先在当前线程的栈帧中创建用于存储锁记录的空间(LockRecord)
将对象头中的Mark Word复制到锁记录中，称为Displaced Mark Word.
线程尝试使用CAS将对象头中的Mark Word替换为指向锁记录的指针
如果替换成功，表示当前线程获得轻量级锁，如果失败，表示存在其他线程竞争锁，那么当前线程会尝试使用CAS来获取锁，当自旋超过指定次数(可以自定义)时仍然无法获得锁，此时锁会膨胀升级为重量级锁

轻量锁解锁

尝试CAS操作将所记录中的Mark Word替换回到对象头中
如果成功，表示没有竞争发生
如果失败，表示当前锁存在竞争，锁会膨胀成重量级锁

一旦锁升级成重量级锁，就不会再恢复到轻量级锁状态。当锁处于重量级锁状态，其他线程尝试获取锁时，都会被阻塞，也就是 BLOCKED状态。当持有锁的线程释放锁之后会唤醒这些现场，被唤醒之后的线程会进行新一轮的竞争

重量级锁

重量级锁依赖对象内部的monitor锁来实现，而monitor又依赖操作系统的MutexLock（互斥锁）

大家如果对MutexLock有兴趣，可以抽时间去了解，假设Mutex变量的值为1，表示互斥锁空闲，这个时候某个线程调用lock可以获得锁，而Mutex的值为0表示互斥锁已经被其他线程获得，其他线程调用lock只能挂起等待

为什么重量级锁的开销比较大呢？

原因是当系统检查到是重量级锁之后，会把等待想要获取锁的线程阻塞，被阻塞的线程不会消耗CPU，但是阻塞或者唤醒一个线程，都需要通过操作系统来实现，也就是相当于从用户态转化到内核态，而转化状态是需要消耗时间的

总结

到目前为止，我们分析了synchronized的使用方法、以及锁的存储、对象头、锁升级的原理。如果有问题，可以扫描二维码留言

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Java并发编程之可重入锁ReentrantLock

发表于 2021-11-29

这是我参与11月更文挑战的第24天，活动详情查看：2021最后一次更文挑战

ReentrantLock概述

ReentrantLock是可重入的独占锁，同时只能有一个线程可以获得锁，其他获取该锁的线程会被阻塞而被放入该锁的AQS阻塞队列里面。我们先看一下ReentrantLock的类图。ReentrantLock其实也是根据AQS来实现的，并且根据的输入参数来决定其是公平锁还是非公平锁，默认是非公平锁。Sync类直接继承自AQS，它的子类NonfairSync和FairSync
分别实现了获取锁的非公平与公平策略。

在AQS的state状态值表示线程获取该锁的可重入次数，在默认情况下，state的值为0表示当前锁没有被任何线程持有。当一个线程第一次获取到该锁时会尝试使用CAS设置state值为1，如果CAS成功则当前线程获取了该锁，然后记录锁的持有者为当前线程。在该线程没有释放锁的情况下，第二次获取到锁时，state被修改为2，这就是可重入次数。在该线程释该锁时，会尝试使用CAS让状态值减1，如果减1后状态为0，则释放掉该锁。

ReentrantLock重要方法详解

void lock()方法

1
2
3

java复制代码public void lock() {
    sync.lock();
}

如上代码所示，ReentrantLock的lock()函数是委托给了Sync类，根据ReentrantLock的构造函数选择sync的实现时NonfailSync还是FairSync。我们接下来先看一下Sync子类NonfairSync的实现。

java复制代码static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    final void lock() {
    //CAS设置状态值
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1); //调用AQS的acqiure方法
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

在NonfairSync代码中，因为默认的state值为0，所以第一个调用Lock的线程会通过CAS设置状态值为1，CAS成功则表示当前线程获取到了锁，通过setExclusiveOwnerThread设置锁的持有者是当前线程。

如果这时候有其他线程调用lock方法企图获取该锁，CAS会失败，然后会调用AQS的acquire方法。(acquire传入的值为1)

java复制代码public final void acquire(int arg) {
//调用ReentrantLock重写的tryAcquire方法
//tryAcquire(arg) 如果为false会把当前线程放入AQS阻塞队列
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

我们之前的文章说过，AQS并没有提供tryAcquire方法，我们先看一下非公平锁的实现。

java复制代码final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    //获取当前锁的state
    int c = getState();
    //如果当前AQS状态为0
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    //如果当前线程是锁的持有者
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

首先会检查锁的state是否为0，如果是则直接CAS抢锁,然后返回true。如果state不为0，则锁已经被某个线程持有，先判断当前线程是否是锁的持有者，如果是则state+1(nextc<0说明可重入次数溢出了)，如果不是则返回false。

然后我们再来看一下，FairSync重写的tryAcquire方法是怎样来实现公平的。

java复制代码protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if  (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

接下来的很重要

公平锁的tryAcquire策略与非公平类似，不同之处在于在设置CAS之前添加了hasQueuePredecessors方法。

java复制代码public final boolean hasQueuedPredecessors() {

    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

在上面代码中，返回值为false的时候，意味着当前节点不用去排队，可以分为两种情况。

(h!=t)为false,则直接返回false。这句话的意思是首尾节点相等，当首尾节点都为null是，说明队列为空。所以不用排队；当首尾节点不为null且相等时，说明队列中只有一个节点，当队列中只有一个节点时，根据我们之前AQS中的对acquireQueued()函数的介绍，第二个节点是不参与排队的。
当(h!=t)返回true时，表示队列中至少有两个不同节点存在；(s = h.next) == null返回false表示头节点有后继节点；s.thread != Thread.currentThread()返回false表示第二个节点的线程就是当前线程，所以也无需排队。

返回值为true的时候，意味着当前节点需要去排队

当(h!=t)返回true时，表示队列中节点数>=2，此时继续判断当((s=h.next)==null)为false时，继续判断(s.thread !=Thread.currentThread())是否为true，如果为true，意思就是头结点有后继节点，并且后继节点不是当前线程，所以当前线程需要去队列中。

void lockInterruptibly()方法

该方法与lock()方法类似，不同点在于它对中断进行响应。当前线程在调用该方法时，如果其它线程调用了当前线程的interrupt()方法，则当前线程会抛出InterruptedException异常。

1
2
3

java复制代码public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
}

java复制代码public final void acquireInterruptibly(int arg)
        throws InterruptedException {
        //如果当前线程被中断，则直接抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
        //尝试获取资源
    if (!tryAcquire(arg))
    //调用AQS可被中断的方法
        doAcquireInterruptibly(arg);
}

boolean tryLock()方法

表示尝试获取锁，如果当前该锁没有被其他线程持有，则当前线程获取该锁并返回true，否则返回false。本方法不会引起当前线程阻塞。

1
2
3

java复制代码public boolean tryLock() {
    return sync.nonfairTryAcquire(1);
}

java复制代码final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

void unlock()方法

尝试释放锁，如果当前线程持有锁，则调用该方法会让该线程对该线程持有的AQS中的state值减1，如果减去1后，state=0,则当前线程释放该锁，否则只是次数state-1。如果线程没有持有该锁，则会抛出IllegalMonitorStateException异常。

1
2
3

java复制代码public void unlock() {
    sync.release(1);
}

java复制代码public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

java复制代码protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

使用redis生成唯一编号

发表于 2021-11-29

这是我参与11月更文挑战的第9天，活动详情查看：2021最后一次更文挑战」
在项目开发中，我们需要保证数据的唯一性，就目前开发中常用的方式有使用自增序列、GUID、时间戳以及时间戳加上随机数。生成ID的方法有很多，每种适用场景、需求以及性能要求不同。下面我们列出以下较为常用的生成ID的方式，并且来讨论以下他们的优缺点。

利用数据库自带的自增功能设置唯一id：

优点：可控并且显而易见。
缺点：对于单库单表来说数据库压力大，对于单库多表来说，id并不是全库唯一。

利用GUID：
生成GUID是长度为32的16进制字符串，如果转换为byte数组则一共有16个byte元素，也就是说GUID是一个128bit长的数字。

优点：减轻了数据库的压力。
缺点：对于需要排序来说，我们无法使用id排序。

TIP：目前有类似GUID的方式，但是大多数都是把时间拼接上去，但是这样就造成了id特别长。

自定义ID：
目前推特使用的是自己开发的全局唯一ID生成服务Snowflake。它是由精确到毫秒的41位时间序列和10位机器标识以及12位的计数顺序号组成的，它的最高位是符号位并且始终为0。

优点：高性能、低延迟、可以按时间排序。
缺点：需要独立开发和部署。

Redis生成id：
对于大型系统来说，我们可以使用Redis来生成ID，主要是依赖于redis是单线程的，因此可以用来生成全局唯一ID。要实现这个功能我们可以用redis的原子操作INCR和INCRBY来实现。下面我们就来看一下如何使用redis生成唯一ID，主要思想是利用redis单线程特性以保证操作的原子性，这样读写同一key时不会出现不同的数据。代码如下：

首先我们先利用DequeueItemFromList方法循环获取编号GetForeachNumbers，

csharp复制代码 private string GetForeachNumbers(IRedisClient redisClent, ShortNumberType type, int count = 5)
 {
     var number = string.Empty;
     var key = string.Format(CacheKeys.ComShortNumberList, type);
     for (var i = 0; i < count; i++)
     {
         number = redisClent.DequeueItemFromList(key);
         if (string.IsNullOrWhiteSpace(number) || string.IsNullOrEmpty(number))
         {
             if (!RedisUpload(type, true))
             {
                 Thread.Sleep(500);
             }
         }
         else
         {
             break;
         }
     }
     return number;
 }

然后使用redis上传序列

csharp复制代码  public bool RedisUpload(ShortNumberType type, bool autoInsert)
  {
      var result = false;
      var key = string.Format(CacheKeys.ComShortNumberList, type);
      using (var redisClent = RedisManager.GetClient())
      {
          if (redisClent.GetListCount(key) <= RedisMinCount)
          {
              var lockKey = "comShortNumber" + type;
              string token;
              if (RedisManager.Lock(lockKey, out token, 1500))
              {
                  try
                  {
                      var list = GetListForRedis(type);

                      //存储量不足，自动新增
                      if (list.Count < RedisUploadCount && autoInsert && AutoInsertList(type))
                      {
                          list = GetListForRedis(type);
                      }
                      if (list.Any())
                      {
                          var ids = list.Select(x => x.Id).ToList();
                          var numbers = list.Select(x => x.Number).ToList();
                          UpdateListByStatu(ids, ShortNumberStatus.handleIng);
                          redisClent.AddRangeToList(key, numbers.OrderByDescending(x => x).ToList());
                          UpdateListByStatu(ids, ShortNumberStatus.Finished);
                      }
                  }
                  finally
                  {
                      RedisManager.DelLock(lockKey, token);
                  }

                  result = true;
              }
          }
          else
              result = true;
      }
      return result;
}

最后获取编号

csharp复制代码public string GetNumber(ShortNumberType type)
{
	using (var redisClent = RedisManager.GetClient())
	{
		return GetForeachNumbers(redisClent, type);
	}
}

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

SpringBoot Actuator 潜在的 OOM 问题

发表于 2021-11-29

这是我参与11月更文挑战的第 14 天，活动详情查看：2021最后一次更文挑战

此问题背景产生于近期需要上线的一个功能的埋点；主要表现就是在应用启动之后的一段时间内，内存使用一直呈现递增趋势。

下图为场景复线后，本地通过 jconsole 查看到的内部使用走势图。

实际环境受限于配置，内存不会膨胀

背景&问题

应用 a 使用 rest template 通过 http 方式调用应用 b，应用项目中开启了 actuator，api 使用的是 micrometer；在 client 调用时，actuator 会产生一个 name 为 http.client.requests 的 metrics，此 metric 的 tag 中包含点目标的 uri。

应用 b 提供的接口大致如下：

java复制代码@RequestMapping("test_query_params")
public String test_query_params(@RequestParam String value) {
    return value;
}

@RequestMapping("test_path_params/{value}")
public String test_path_params(@PathVariable String value) {
    return value;
}

http://localhost:8080/api/test/test_query_params?value=

http://localhost:8080/api/test/test_path_params/{value}_

期望在 metric 的收集结果中应该包括两个 metrics，主要区别是 tag 中的 uri 不同，一个是 api/test/test_query_params，另一个是 api/test/test_path_params/{value}；实际上从拿到的 metrics 数据来看，差异很大，这里以 pathvariable 的 metric 为例，数据如下：

json复制代码tag: "uri",
values: [
"/api/test/test_path_params/glmapper58",
"/api/test/test_path_params/glmapper59",
"/api/test/test_path_params/glmapper54",
"/api/test/test_path_params/glmapper55",
"/api/test/test_path_params/glmapper56",
"/api/test/test_path_params/glmapper57",
"/api/test/test_path_params/glmapper50",
"/api/test/test_path_params/glmapper51",
"/api/test/test_path_params/glmapper52",
"/api/test/test_path_params/glmapper53",
"/api/test/test_path_params/glmapper47",
"/api/test/test_path_params/glmapper48",
"/api/test/test_path_params/glmapper49",
"/api/test/test_path_params/glmapper43",
"/api/test/test_path_params/glmapper44",
"/api/test/test_path_params/glmapper45",
"/api/test/test_path_params/glmapper46",
"/api/test/test_path_params/glmapper40",
"/api/test/test_path_params/glmapper41",
"/api/test/test_path_params/glmapper42",
"/api/test/test_path_params/glmapper36",
"/api/test/test_path_params/glmapper37",
"/api/test/test_path_params/glmapper38",
"/api/test/test_path_params/glmapper39",
"/api/test/test_path_params/glmapper32",
"/api/test/test_path_params/glmapper33",
"/api/test/test_path_params/glmapper34",
"/api/test/test_path_params/glmapper35",
"/api/test/test_path_params/glmapper30",
"/api/test/test_path_params/glmapper31",
"/api/test/test_path_params/glmapper25",
"/api/test/test_path_params/glmapper26",
....
]

可以非常明显的看到，这里将{value} 参数作为了 uri 组件部分，并且体现在 tag 中，并不是期望的 api/test/test_path_params/{value}。

问题原因及解决

两个问题，1、这个埋点是怎么生效的，先搞清楚这个问题，才能顺藤摸瓜。2、怎么解决。

默认埋点是如何生效的

因为是通过 resttemplate 进行调用访问，那么埋点肯定也是基于对 resttemplate 的代理；按照这个思路，笔者找到了 org.springframework.boot.actuate.metrics.web.client.MetricsRestTemplateCustomizer 这个类。RestTemplateCustomizer 就是对 resttemplate 进行定制的，MetricsRestTemplateCustomizer 通过名字也能得知期作用是为了给 resttemplate 增加 metric 能力。

再来讨论 RestTemplateCustomizer，当使用RestTemplateBuilder构建RestTemplate时，可以通过RestTemplateCustomizer进行更高级的定制，所有RestTemplateCustomizer beans 将自动添加到自动配置的RestTemplateBuilder。也就是说如果想 MetricsRestTemplateCustomizer 生效，那么构建 resttemplate 必须通过 RestTemplateBuilder 方式构建，而不是直接 new。

http.client.requests 中的 uri

塞 tag 的代码在org.springframework.boot.actuate.metrics.web.client.RestTemplateExchangeTags 类中，作用时机是在 MetricsClientHttpRequestInterceptor 拦截器中。当调用执行完成后，会将当次请求 metric 记录下来，在这里就会使用到 RestTemplateExchangeTags 来填充 tags。下面仅给出 uri 的部分代码

java复制代码
	/**
	 * Creates a {@code uri} {@code Tag} for the URI of the given {@code request}.
	 * @param request the request
	 * @return the uri tag
	 */
	public static Tag uri(HttpRequest request) {
		return Tag.of("uri", ensureLeadingSlash(stripUri(request.getURI().toString())));
	}

	/**
	 * Creates a {@code uri} {@code Tag} from the given {@code uriTemplate}.
	 * @param uriTemplate the template
	 * @return the uri tag
	 */
	public static Tag uri(String uriTemplate) {
		String uri = (StringUtils.hasText(uriTemplate) ? uriTemplate : "none");
		return Tag.of("uri", ensureLeadingSlash(stripUri(uri)));

其余的还有 status 和 clientName 等 tag name。

通过断点，可以看到，这里 request.getURI() 拿到的是带有参数的完整请求链接。

这些 tag 的组装最终在 DefaultRestTemplateExchangeTagsProvider 中完成，并返回一个列表。

java复制代码private Timer.Builder getTimeBuilder(HttpRequest request, ClientHttpResponse response) {
    return this.autoTimer.builder(this.metricName)
                // tagProvider 为 DefaultRestTemplateExchangeTagsProvider
				.tags(this.tagProvider.getTags(urlTemplate.get().poll(), request, response))
				.description("Timer of RestTemplate operation");
}

解决

这里先来看下官方对于 request.getURI 的解释

java复制代码
	/**
	 * Return the URI of the request (including a query string if any,
	 * but only if it is well-formed for a URI representation).
	 * @return the URI of the request (never {@code null})
	 */
	URI getURI();

返回请求的 URI，这里包括了任何的查询参数。那么是不是拿到不用参数的 path 就行呢？

这里尝试通过 request.getURI().getPath() 拿到了预期的 path（@pathvariable 拿到的是模板）。

再回到 DefaultRestTemplateExchangeTagsProvider，所有的 tag 都是在这里完成组装，这个类明显是一个默认的实现(Spring 体系下基本只要是Defaultxxx 的，一般都能扩展 )，查看它的接口类 RestTemplateExchangeTagsProvider 如下：

java复制代码
/**
 * Provides {@link Tag Tags} for an exchange performed by a {@link RestTemplate}.
 *
 * @author Jon Schneider
 * @author Andy Wilkinson
 * @since 2.0.0
 */
@FunctionalInterface
public interface RestTemplateExchangeTagsProvider {

	/**
	 * Provides the tags to be associated with metrics that are recorded for the given
	 * {@code request} and {@code response} exchange.
	 * @param urlTemplate the source URl template, if available
	 * @param request the request
	 * @param response the response (may be {@code null} if the exchange failed)
	 * @return the tags
	 */
	Iterable<Tag> getTags(String urlTemplate, HttpRequest request, ClientHttpResponse response);

}

RestTemplateExchangeTagsProvider 的作用就是为 resttemplate 提供 tag 的，所以这里通过自定义一个 RestTemplateExchangeTagsProvider，来替换DefaultRestTemplateExchangeTagsProvider，以达到我们的目标，大致代码如下：

java复制代码
@Override
 public Iterable<Tag> getTags(String urlTemplate, HttpRequest request, ClientHttpResponse response) {
    Tag uriTag;
    // 取 request.getURI().getPath() 作为 uri 的 value
    if (StringUtils.hasText(request.getURI().getPath())) {
      uriTag = Tag.of("uri", ensureLeadingSlash(stripUri(request.getURI().getPath())));
    } else {
      uriTag = (StringUtils.hasText(urlTemplate) ? RestTemplateExchangeTags.uri(urlTemplate)
                    : RestTemplateExchangeTags.uri(request));
    }
    return Arrays.asList(RestTemplateExchangeTags.method(request), uriTag,
                RestTemplateExchangeTags.status(response), RestTemplateExchangeTags.clientName(request));
    }

会不会 OOM

理论上，应该参数不同，在使用默认 DefaultRestTemplateExchangeTagsProvider 的情况下，meter 会随着 tags 的不同迅速膨胀，在 micrometer 中，这些数据是存在 map 中的

1
2
3

java复制代码// Even though writes are guarded by meterMapLock, iterators across value space are supported
// Hence, we use CHM to support that iteration without ConcurrentModificationException risk
private final Map<Id, Meter> meterMap = new ConcurrentHashMap<>();

一般情况下不会，这里是因为 spring boot actuator 自己提供了保护机制，对于默认情况，tags 在同一个 metric 下，最多只有 100 个

java复制代码/**
* Maximum number of unique URI tag values allowed. After the max number of
* tag values is reached, metrics with additional tag values are denied by
* filter.
*/
private int maxUriTags = 100;

如果你想使得这个数更大一些，可以通过如下配置配置

1	properties复制代码management.metrics.web.client.max-uri-tags=10000

如果配置值过大，会存在潜在的 oom 风险。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

linux中centos7防火墙开启/关闭异常错误解决方案

发表于 2021-11-29

「这是我参与11月更文挑战的第28天，活动详情查看：2021最后一次更文挑战」。

根本报错信息：Failed to start firewalld.service: Unit is masked.（防火墙启动不了）

centos7介绍

使用背景：

1，CentOS7 默认的防火墙不是iptables, 而是firewalle.

2，端口号使用需要单独放开来提供服务，如：8080，80等常用端口

系统、环境

问题起因：在idea集成docker，连接虚拟机路径测试时，出现以下连接异常提示信息。

解决过程：

发现问题一：查看防火墙状态： systemctl status firewalld，发现防火墙没有开启。

解决方案一：一般情况，开启防火墙后，问题就解决了，然而当开启防火墙时，发现防火墙也开启不了，开启防火墙命令：systemctl start firewalld

问题二：防火墙开启失败

解决方案二：systemctl unmask firewalld

问题可能是上次异常关闭，导致有残留的文件妨碍了防火墙的开启，需要清除原来的firewalld残留文件，然后就可以用systemctl start firewalld来启动防火墙了。

问题三：(可能没有设置端口号，一般为：2375)

解决方案三：

需要在dicker。service下设置docker的端口号，然后再开放端口号，最后需要重启防火墙就可以正常使用了。

设置端口号（vim /lib/systemd/system/docker.service）

重启防火墙（firewall-cmd –reload）

效果：

附加：接下来记录一下idea如何集成docke，一键打包部署微服务

所谓idea集成docker，首先就是先开启docker远程访问。

1、如何开启docker远程访问？

1、
修改该Docker服务文件

1）进入：docker.service —命令：vi /lib/systemd/system/docker.service
2）修改：ExecStart这行，开启远程访问路径权限—
ExecStart=/usr/bin/dockerd -H tcp://0.0.0.0:2375 -H unix:///var/run/docker.sock

默认是：（ExecStart=/usr/bin/dockerd -H fd:// –containerd=/run/containerd/containerd.sock）

2、
重新开启，确认。

#重新加载配置文件

systemctl daemon-reload

#重启服务

systemctl restart docker.service

#查看端口是否开启

netstat -nlpt #如果找不到netstat命令，可进行安装：yum install net-tools

#直接curl看是否生效

curl http://127.0.0.1:2375/info

2、IDEA安装Docker插件

打开Idea，从File->Settings->Plugins->Install JetBrains plugin进入插件安装界面，点击install安装，安装后重启Idea。

3、IDEA配置Docker

从File->Settings->Build,Execution,Deployment->Docker打开配置界面，即上文出现问题的情况，解决后的效果。

配置docker，连接到远程docker服务。

4、docker-maven-plugin

docker-maven-plugin 插件就是为了帮助我们在Maven工程中，通过简单的配置，自动生成镜像并推送到仓库中。在pom.xml中进行配置

xml复制代码<properties>
     <docker.image.prefix>name</docker.image.prefix>
</properties>

 <build>
     <plugins>
         <plugin>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-maven-plugin</artifactId>
             <version>1.0.0</version>
         </plugin>
         
          <plugin>
                <groupId>com.spotify</groupId>
                <artifactId>docker-maven-plugin</artifactId>
                <version>1.0.0</version>

                <configuration>
                    <!-- 镜像名称  name/exam-->
                    <imageName>${docker.image.prefix}/${project.artifactId}</imageName>
                    <!--指定标签-->
                    <imageTags>
                        <imageTag>latest</imageTag>
                    </imageTags>
                    <!-- 基础镜像jdk 1.8-->
                    <baseImage>java</baseImage>
                    <!-- 制作者提供本人信息 -->
                    <maintainer>name@aliyun.com</maintainer>
                    <!--切换到/ROOT目录 -->
                    <workdir>/ROOT</workdir>
                    <cmd>["java", "-version"]</cmd>
                    <entryPoint>["java", "-jar", "${project.build.finalName}.jar"]</entryPoint>

                    <!-- 指定 Dockerfile 路径
                      <dockerDirectory>${project.basedir}/src/main/docker</dockerDirectory>
                   -->

                    <!--指定远程 docker api地址-->
                    <dockerHost>http://0.0.0.0:2375</dockerHost>

                    <!-- 这里是复制 jar 包到 docker 容器指定目录配置 -->
                    <resources>
                        <resource>
                            <targetPath>/ROOT</targetPath>
                            <!--用于指定需要复制的根目录，${project.build.directory}表示target目录-->
                            <directory>${project.build.directory}</directory>
                            <!--用于指定需要复制的文件。${project.build.finalName}.jar指的是打包后的jar包文件。-->
                            <include>${project.build.finalName}.jar</include>
                        </resource>
                    </resources>
                </configuration>

            </plugin>
         
         
     </plugins>
</build>

完成以上操作，就可以之间用maven打包，启动镜像，开启容器。进行docker操作了。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Oracle TX 锁快速查杀处理办法

发表于 2021-11-29

根据保护的数据不同，ORACLE的数据库锁分为以下几大类：

1.DML锁（data locks数据锁），用于保护数据的完整性；

2.DDL锁（dictionary locks字典锁），用于保护数据库对象的结构，例如表、索引的结构定义；

3.内部锁或闩（internal locks or latches），用于保护内部结构

在 Oracle 数据库中，DML锁主要包括 TM 锁和 TX 锁，其中 TM 锁称为表级锁，TX 锁称为事务锁或行级锁。

当 Oracle 执行 DML 语句时，系统自动在所要操作的表上申请 TM 类型的锁。当 TM 锁获得后，系统再自动申请 TX 类型的锁，并将实际锁定的数据行的锁标志位进行置位。这样在事务加锁前检查 TX 锁相容性时就不用再逐行检查锁标志，而只需检查 TM 锁模式的相容性即可，大大提高了系统的效率。TM 锁包括了 SS、SX、S、X 等多种模式，在数据库中用 0－6 来表示。不同的 SQL 操作产生不同类型的 TM 锁。

在数据行上只有 X 锁（排他锁）。在 Oracle 数据库中，当一个事务首次发起一个 DML 语句时就获得一个 TX 锁，该锁保持到事务被提交或回滚。当两个或多个会话在表的同一条记录上执行 DML 语句时，第一个会话在该条记录上加锁，其他的会话处于等待状态。当第一个会话提交后，TX 锁被释放，其他会话才可以加锁。

当 Oracle 数据库发生 TX 锁等待时，如果不及时处理常常会引起 Oracle 数据库挂起，或导致死锁的发生，产生 ORA-60 的错误。这些现象都会对实际应用产生极大的危害，如长时间未响应，大量事务失败等。

ORACLE里锁有以下几种模式:

**0：none

1：null 空

2：Row-S 行共享(RS)：共享表锁，sub share

3：Row-X 行独占(RX)：用于行的修改，sub exclusive

4：Share 共享锁(S)：阻止其他DML操作，share

5：S/Row-X 共享行独占(SRX)：阻止其他事务操作，share/sub exclusive

6：exclusive 独占(X)：独立访问使用，exclusive**

上面提到 TX 锁也称事务锁或者行级锁，是控制数据库并发访问的一项重要技术，也是数据完整性和一致性的重要保证。下面就一起看看如何快速查杀锁。

方案 1

csharp复制代码--行锁查看

col LOCK_ID1 for a10
col "Locked Mode" for a15
col MACHINE for a20 
set line 345 pages 345
col "Kill" for a40 
select s.SID,s.SERIAL#,s.MACHINE,s.TYPE,l.TYPE,l.CTIME,l.BLOCK,l.REQUEST,l.LMODE,
       decode(l.lmode,
              0,
              'None',
              1,
              'Null',
              2,
              'Row-S (SS)',
              3,
              'Row-X (SX)',
              4,
              'Share',
              5,
              'S/Row-X (SSX)',
              6,
              'Exclusive',
              substr(to_char(l.lmode), 1, 13)) as "Locked Mode",
       DECODE(L.TYPE,
              'MR',
              'File_ID:' || L.ID1,
              'TM',
              t.NAME,
              'TX',
              'USN:' || to_char(TRUNC(L.ID1 / 65536)) || 'RWO:' ||
              nvl(r.NAME, 'None'),
              L.ID1) as LOCK_ID1,
       'alter system kill session ''' || s.SID || ',' || s.SERIAL# || '''immediate;' as "Kill"
  from gv$process p
 inner join gv$session s
    on s.PADDR = p.ADDR
 inner join v$lock l
    on l.SID = s.SID
  left join sys.obj$ t
    on l.ID1 = t.obj#
  left join sys.obj$ r
    on s.ROW_WAIT_OBJ# = r.obj#
 where 1 = 1
   and l.TYPE != 'MR' 
 and l.TYPE  = 'TM'
   and l.lmode = 3
 order by s.SID;

通过上方 SQL 可定位锁以及查杀语句，但无法确定锁源头，可通过如下 SQL 查到 SID 定位锁源头信息，如查出的锁源头（SID 位于最左边）在上边结果集中出现，则可使用上面 SQL 直接查杀锁，方便快速。

sql复制代码set linesize 132
col sid format a12
col event format a30
col sql_text format a40
col object_name format a25
WITH sessions AS
(SELECT  /*+materialize*/
         sid
        ,sql_id
        ,event
        ,blocking_session
        ,row_wait_obj#
FROM gv$session)
SELECT LPAD(' ', LEVEL ) || sid sid
      ,sql_id
      ,event
      ,owner||decode(owner,null,null,'.')||object_name object_name
      ,substr(sql_text,1,40) sql_text
FROM sessions s
LEFT OUTER JOIN dba_objects ON (object_id = row_wait_obj#)
LEFT OUTER JOIN v$sql USING (sql_id)
WHERE sid IN (SELECT blocking_session FROM sessions)
OR blocking_session IS NOT NULL
CONNECT BY PRIOR sid = blocking_session
   START WITH blocking_session IS NULL;

方案 2

使用 v$session 中的 final_blocking_instance 和 final_blocking_session 定位锁源头，具体 SQL 如下：

sql复制代码SQL> select 'alter system kill session ''' || ss.sid || '' || ',' || ss.serial# || ',@' ||
       ss.inst_id || ''' immediate;' db_kill_session
  from gv$session s, 
  gv$session ss
 where s.final_blocking_session is not null
   and s.final_blocking_instance = ss.inst_id
   and s.final_blocking_session = ss.sid
   and s.sid <> ss.sid;
DB_KILL_SESSION
--------------------------------------------------
alter system kill session '161,5579,@1' immediate;
alter system kill session '161,5579,@1' immediate;

--如下，可通过操作系统层面直接 kill 查杀锁。
SQL> select p.inst_id, 'kill -9 ' || p.spid os_kill_session
  from gv$session s, 
  gv$session ss, 
  gv$process p
 where s.final_blocking_session is not null
   and s.final_blocking_instance = ss.inst_id
   and s.final_blocking_session = ss.sid
   and ss.paddr = p.addr
   and ss.inst_id = p.inst_id
   and s.sid <> ss.sid;
   INST_ID OS_KILL_SESSION
---------- --------------------------------
         1 kill -9 12349

最后在分享一些通用的杀会话的 SQL 脚本：

python复制代码1、kill某个等待事件对应的spid：
set linesize 260 pagesize 10000
select 'kill -9 ' || a.spid
  from v$process a, v$session b
 where a.addr = b.paddr
   and a.background is null
   and b.type = 'USER'
   and b.event like '%' || '&eventname' || '%'
   and b.status = 'ACTIVE';

-- 对应的alter system kill session的语法：

set linesize 260 pagesize 1000
col machine for a50
col kill_session for a60;
select machine,
       'alter system kill session ' || ''''||sid|| ',' || serial# ||''''|| 'immediate;' kill_session,
       status
  from v$session
 where type='USER' and event like '%event_name%' and status = 'ACTIVE';

2、kill某个sql_id对应的spid：
set linesize 260 pagesize 10000
select 'kill -9 ' || a.spid
  from v$process a, v$session b
 where a.addr = b.paddr
   and a.background is null
   and b.type = 'USER'
   and b.sql_id = '&sql_id'
   and b.status = 'ACTIVE';

-- 对应的alter system kill session的语法：

set linesize 260 pagesize 10000
col machine for a60
select machine,
       'alter system kill session ' || ''''||sid|| ',' || serial# ||''''|| 'immediate;',
       status
  from v$session
 where sql_id = '&sql_id' and type='USER' and status='ACTIVE';

3、被kill会话的类型：
set linesize 260 pagesize 10000
select b.osuser,b.machine,b.program,b.sql_id,b.PREV_SQL_ID,a.spid,to_char(LAST_CALL_ET) as seconds,b.BLOCKING_SESSION,b.BLOCKING_INSTANCE
  from v$process a, v$session b
 where a.addr = b.paddr
   and a.inst_id=b.inst_id
   and a.background is null
   and b.type = 'USER'
   and b.event='&event_name'
   and b.status = 'ACTIVE';

4、blocking会话类型和kill blocking会话：
set linesize 260 pagesize 10000
col machine for a50
col kill_session for a60
SELECT
    blocking_instance,
    blocking_session,
    BLOCKING_SESSION_STATUS,
    FINAL_BLOCKING_INSTANCE,
    FINAL_BLOCKING_SESSION,
    COUNT(*)
FROM
    v$session
WHERE
    upper(event) LIKE '%&cursor%'
GROUP BY
    blocking_instance,
    blocking_session,
    BLOCKING_SESSION_STATUS,
    FINAL_BLOCKING_INSTANCE,
    FINAL_BLOCKING_SESSION
    order by blocking_instance,count(*);

-- kill blocking会话

select 
       inst_id,
       machine,
       'alter system kill session ' || ''''||sid|| ',' || serial# ||''''|| 'immediate;' kill_session,
       status
  from gv$session a
where a.type='USER' and (a.inst_id,a.sid) in 
(
select 
BLOCKING_INSTANCE,
BLOCKING_SESSION 
from v$session
where upper(event) like '%&cursor%'
)
order by inst_id;

5、所有含有关键字“LOCAL=NO”的进程是Oracle数据库中远程连接进程的共同特点，因此通过以下命令可以kill掉所有的进程
ps -ef|grep -v grep|grep LOCAL=NO|awk '{print $2}'|xargs kill -9

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Java集合学习(七)——深入理解LinkedHashSet

发表于 2021-11-29

这是我参与11月更文挑战的第27天，活动详情查看：2021最后一次更文挑战」

前言

1	复制代码大家好，我是程序猿小白 GW_gw，很高兴能和大家一起学习进步。

在读本篇之前希望读者先了解HashSet类，这样对与LinkedHashSet类的学习会更加容易。可以参考以下链接：

Java集合学习(六)——深入理解HashSet类

以下内容部分来自于网络，如有侵权，请联系我删除，本文仅用于学习交流，不用作任何商业用途。

摘要

1	复制代码本文主要介绍Java集合的LinkedHashSet类的基本特点和常用的方法。

1. LinkedHashSet类

1.1 什么是LinkedHashSet类

LinkedHashSet继承了HashSet。与HashMap有所不同的是它底层使用LinkedHashMap来实现，即在HashMap的基础上新增加了一个双向链表来确保迭代有序(迭代的元素顺序和插入的元素顺序相同)。因此性能略低于HashSet，但是迭代访问元素时将具有很好的性能，因为它的内部是以链表来实现。

允许存放null元素，此实现不是同步的。

1.2 LinkedHashSet类的构造方法

LinkedHashSet有四个构造方法，如下：

LinkedHashSet() 构造一个带默认初始容量 (16) 和加载因子 (0.75) 的新空链接哈希 set。
LinkedHashSet(Collection c) 构造一个与指定 collection 中的元素相同的新链接哈希 set。
LinkedHashSet(int initialCapacity) 构造一个带指定初始容量和默认加载因子 (0.75) 的新空链接哈希 set。
LinkedHashSet(int initialCapacity, float loadFactor) 构造一个带有指定初始容量和加载因子的新空链接哈希 set。

实例展示：

ini复制代码public class LinkedHashSetTest {
    public static void main(String[] args) {
        LinkedHashSet linkedHashSet = new LinkedHashSet<>();
        ArrayList arrayList = new ArrayList<String>();
        arrayList.add("1");
        arrayList.add("2");
        arrayList.add("3");
        LinkedHashSet linkedHashSet1 = new LinkedHashSet<>(arrayList);
        LinkedHashSet linkedHashSet2 = new LinkedHashSet<>(32);
        LinkedHashSet linkedHashSet3 = new LinkedHashSet<>(32,0.8f);
        linkedHashSet2.add(1);
        linkedHashSet2.add(3);
        linkedHashSet2.add(2);
        linkedHashSet2.add(0);
        Iterator iterator = linkedHashSet2.iterator();
        while(iterator.hasNext()){
            Object next = iterator.next();
            System.out.println(next);
        }
        
    }
}

可以看到，输出元素的顺序和插入元素的顺序是一致的，这是LinkedHashSet类的一大特点。

1.3 LinkedHashSet类的普通方法

LinkedHashSet并没有自己独有的方法，LinkedHashSet的方法都是从父类HashSet继承和实现Set接口和Object类而来，这里就不再赘诉。

小结

以上就是LinkedHashSet类的一些介绍，到此我们已经把Set集合中的一些常用类学完了，接下来我们继续来学习Collection集合中的List集合。以上内容希望对读者有所帮助，如有不足之处，欢迎留言指正。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

可能是最好的Spring RMI开发教程【建议收藏】

发表于 2021-11-29

「这是我参与11月更文挑战的第31天，活动详情查看：2021最后一次更文挑战」。

前言

Spring 作为一个基础框架，对 remoting（远程调用）有多种支持。Spring remoting support 简化了远程服务的开发。

目前，Spring 支持以下远程技术：Remote Method Invocation (RMI)、HTTP Invoker、Hessian、Burlap、JAX-RPC、JAX-WS 和 JMS。

Spring支持的远程技术

远程方法调用 (RMI)

Spring 通过 RmiProxyFactoryBean 和 RmiServiceExporter 支持 RMI。 RmiServiceExporter 将任何 Spring 管理的 bean 作为 RMI 服务导出并注册。 RmiProxyFactoryBean 是一个为 RMI 服务创建代理的工厂 bean。该代理对象代表客户端与远程 RMI 服务对话。

Spring的HTTP invoker

Spring HTTP invoker 使用标准的 Java 序列化机制通过 HTTP 暴露服务。 Spring 通过 HttpInvokerProxyFactoryBean 和 HttpInvokerServiceExporter 支持 HTTP 调用者基础设施。 HttpInvokerServiceExporter 将指定的服务 bean 导出为 HTTP 调用程序服务端点，可通过 HTTP 调用程序代理访问。 HttpInvokerProxyFactoryBean 是 HTTP 调用代理的工厂 bean。

Hessian

Hessian 提供了一种基于 HTTP 的二进制远程处理协议。 Spring 通过 HessianProxyFactoryBean 和 HessianServiceExporter 支持 Hessian。

Burlap

Burlap 是 Caucho 基于 XML 的 Hessian 替代品。 Spring 提供了 BurlapProxyFactoryBean 和 BurlapServiceExporter 等支持类。

JAX-RPC

Spring 通过 JAX-RPC（J2EE 1.4 的 Web 服务 API）为 Web 服务提供远程支持。

JAX-WS

Spring 通过 JAX-WS（Java EE 5 和 Java 6 中引入的 JAX-RPC 的继任者）为 Web 服务提供远程支持。

JMS

Spring 中的 JMS 远程处理支持由 JmsInvokerServiceExporter 和 JmsInvokerProxyFactoryBean 类提供。

Spring开发RMI服务

1、创建Maven项目

2、Pom文件添加依赖

xml复制代码<dependencies>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-core</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
<dependencies>

3、添加实体类

java复制代码@Data
public class User implements Serializable {
    private long id;
    private String name;
    private String userName;
}

4、添加存放用户信息的本地缓存接口

1
2
3

java复制代码public interface ICacheService {
    ConcurrentMap<Long, User> getUserMap();
}

5、存放用户信息的本地缓存接口实现类

java复制代码public class CacheServiceImpl implements ICacheService {

    // 使用ConcurrentHashMap存放用户信息
    private ConcurrentHashMap<Long, User> userMap;

    @Override
    public ConcurrentHashMap<Long, User> getUserMap() {
        return userMap;
    }

    public void setUserMap(ConcurrentHashMap<Long, User> userMap) {
        this.userMap = userMap;
    }
}

6、添加用户操作接口

java复制代码public interface IRMIUserService {

    public boolean addUser(User user);

    public boolean deleteUser(User user);

    public List<User> getUserList();
}

7、用户操作接口实现类

java复制代码@Slf4j
public class RMIUserServiceImpl implements IRMIUserService {

    ICacheService cacheService;

    @Override
    public boolean addUser(User user) {
        getCacheService().getUserMap().put(user.getId(), user);
        log.debug("添加用户到缓存，用户信息 = {}", getCacheService().getUserMap().get(user.getId()));
        return true;
    }

    @Override
    public boolean deleteUser(User user) {
        getCacheService().getUserMap().put(user.getId(), user);
        log.debug("用户: {}，已经从缓存删除 " , user);
        return true;
    }

    @Override
    public List<User> getUserList() {
        List<User> list = new ArrayList<>(getCacheService().getUserMap().values());
        log.debug("用户列表 = {}", list);
        return list;
    }

    private ICacheService getCacheService() {
        return cacheService;
    }

    public void setCacheService(ICacheService cacheService) {
        this.cacheService = cacheService;
    }
}

8、RMI服务端

Java复制代码public class RMIServer {
    public static void main(String[] args) {
        new ClassPathXmlApplicationContext("rmiServerAppContext.xml");
    }
}

9、RMI服务端XML配置

xml复制代码<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">

    <bean id="UserMap" class="java.util.concurrent.ConcurrentHashMap" />

    <bean id="CacheService" class="com.rameo.rmi.CacheServiceImpl">
        <property name="userMap" ref="UserMap"/>
    </bean>

    <bean id="RMIUserService" class="com.rameo.rmi.RMIUserServiceImpl" >
        <property name="cacheService" ref="CacheService"/>
    </bean>

    <!-- 声明 RMI Server -->
    <bean class="org.springframework.remoting.rmi.RmiServiceExporter">

        <!-- serviceName 表示 RMI 服务名称 -->
        <property name="serviceName" value="rmiUserService"/>

        <!-- service 表示 RMI 对象 -->
        <property name="service" ref="RMIUserService"/>

        <!-- serviceInterface 表示暴露的 RMI 服务接口 -->
        <property name="serviceInterface" value="com.rameo.rmi.IRMIUserService"/>

        <!-- 注册端口 1099 -->
        <property name="registryPort" value="1099"/>
    </bean>
</beans>

10、RMI客户端

java复制代码@Slf4j
public class RMIClient {
    public static void main(String[] args) {

        log.debug("RMI Service Client is starting...");

        ApplicationContext context = new ClassPathXmlApplicationContext("rmiClientAppContext.xml");

        // 远程用户服务通过 RMI 客户端应用程序上下文调用
        IRMIUserService rmiClient = (IRMIUserService) context.getBean("rmiUserService");

        User user1 = new User();
        user1.setId(1);
        user1.setName("张三");
        user1.setUserName("zhangsan");

        User user2 = new User();
        user1.setId(2);
        user1.setName("李四");
        user1.setUserName("lisi");

        // 通过远程调用添加用户到缓存
        rmiClient.addUser(user1);
        rmiClient.addUser(user2);

        // 通过远程调用从缓存获取用户列表
        rmiClient.getUserList();

        // 通过远程调用从缓存删除用户
        rmiClient.deleteUser(user1);
        rmiClient.deleteUser(user2);

        log.debug("RMI Service Client is stopped...");
    }
}

11、RMI客户端XML配置

xml复制代码<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">

    <!-- 声明 RMI Client -->
    <bean id="rmiUserService" class="org.springframework.remoting.rmi.RmiProxyFactoryBean">
        <!-- serviceUrl 表示 RMI server 调用地址 -->
        <property name="serviceUrl" value="rmi://127.0.0.1:1099/rmiUserService"/>

        <!-- serviceInterface 表示 RMI server 接口调用 -->
        <property name="serviceInterface" value="com.rameo.rmi.IRMIUserService"/>

        <property name="refreshStubOnConnectFailure" value="true"/>
    </bean>
</beans>

12、启动RMI服务端和客户端

启动RMI服务端

启动RMI客户端

13、查看RMI服务端接口被调用的日志

总结

通过调用日志可以看到，服务端通过 RMI registry 暴露了一个服务地址，端口为1099。

客户端通过 rmi 协议，调用 rmi://127.0.0.1:1099/rmiUserService 地址，远程调用服务端接口。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有